Kafka消息中间件

news2025/1/11 23:59:17

Kafka消息中间件

同时市场上也发展处ActiveMq、RabbitMQ、Kafka、RocketMQ、Pulsar等众多优秀的框架；在大数据领域中Kafka目前是使用较多的框架。Kafka作为内部消息通知的框架，可以适应项目中大数据量的高吞吐、实时流计算等功能实现。

分布式消息中间件

在这里插入图片描述

虚拟机配置环境

下载完，tar -zxvf 解压

注意：要使用IP访问kafka需要开放host port（本机的ip），在server.properties下面配置：

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://your.host.name:9092

开启：

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 
jps
bin/kafka-server-start.sh config/server.properties

开启zookeeper可以输入jps看下开启状态，有QuorumPeerMain即可

在这里插入图片描述

依赖

<!-- kafkfa -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <exclusions>
        <exclusion>
            <artifactId>connect-json</artifactId
                <groupId>org.apache.kafka</groupId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

<!-- reflections实现包扫描 -->
<reflections.version>0.9.11</reflections.version>
<dependency>
    <groupId>org.reflections</groupId>
    <artifactId>reflections</artifactId>
    <version>${reflections.version}</version>
</dependency>


<!--json处理-->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <scope>provided</scope>
</dependency>

巧用Reflections库实现包扫描（扫描某个包中某个接口实现、注解等）

作用是扫描没个工程中某一个相同特性的类，一起处理。

    // MyInterface是这些接口的公共特性
    @Test
    public void testReflections() {
        Reflections reflections = new Reflections("要扫描的接口所在的包");
        Set<Class<? extends MyInterface>> classes = reflections.getSubTypesOf(MyInterface.class);

        // 业务操作
        for(Class clazz : classes) {
            //logger.info(clazz.getName());
            System.out.println("Found: " + clazz.getName());
        }
    }

配置

公共模块的配置kafka.properties

# kafka config
kafka.hosts=192.168.169.135:9092

# 内容现在默认接收模块中的多环境配置，也可以自己写内容

# 接收组，这里是项目包名
kafka.group=com.lead.news.${profiles.name}.${spring.application.name}

# 单消息通道，需要以sigle结尾，测试名称
# 多消息通道，自己区别
kafka.topic.admin-test=${kafka.topic.admin-test}

引用模块的配置，发送跟接收都要配置

import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

@Configuration
@ComponentScan("配置处的包名")
public class KafkaConfig {
}

工具配置类

  • KafkaProducerConfig自动配置Kafka消费者
  • KafkaConsumerConfig自动配置Kafka消费者
  • RetryErrorHandler实现消费者处理消息失败后重新发送到消息队列
  • KafkaMessage实现对发送的消息包装，提供重试次数、分类等信息
  • KafkaSender实现消息的统一发送入口功能
  • KafkaTopicConfig自动装载topic名称信息
  • KafkaListener提供自动注册消息消费监听接口类
  • KafkaListenerFactory提供启动时自动注册实现了KafkaListener的消息消费者

在这里插入图片描述

测试

import com.lead.news.configuration.kafka.KafkaListener;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

/**
 * Title：消息消费者
 * Description：
 * @author WZQ
 * @version 1.0.0
 * @date 2020/2/8
 */
@Component
public class TestKafkaListener implements KafkaListener<String,String> {

    /**
     * 消息名称，必须保持一致
     * @return
     */
    @Override
    public String topic () {
        return "topic.test";
    }

    /**
     * 接收消息方法
     * @param data
     * @param consumer
     */
    @Override
    public void onMessage (ConsumerRecord< String, String > data, Consumer< ?, ?> consumer){
        System.out.println("===========receive test message接受消息:" + data);
    }
}

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;

/**
 * Title：消息生产者
 * Description：
 * @author WZQ
 * @version 1.0.0
 * @date 2020/2/8
 */
@SpringBootTest
@RunWith(SpringRunner.class)
public class KafkaTest {

    @Resource
    KafkaTemplate<String, String> kafkaTemplate;

    @Test
    public void test(){
        try {
            // 消息名称，必须保持一致；key可以随便；消息内容
            this.kafkaTemplate.send("topic.test", "123key","123value");
            System.out.println("=================消息发送了！！！================");
            Thread.sleep(5000);// 休眠等待消费者接收消息
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

使用例子

配置模块

配置类kafka.properties中加入新增的消息名称

# article消息名称
kafka.topic.submit-article-auth=${kafka.topic.submit-article-auth}

在工具类中的定义消息名称的KafkaTopicConfig类中加入新增的消息名称变量，获取配置中的名称

//文章提交的kafka消息审核名称
String submitArticleAuth;

消息内容的dto，根据业务

import lombok.Data;

/**
 * Title：封装传递的消息，并且区分消息的类型
 * Description：对应数据库的字段
 * @author WZQ
 * @version 1.0.0
 * @date 2020/2/9
 */
@Data
public class SubmitArticleAuto {

    // 文章类型
    private ArticleType type;
    // 文章ID
    private Integer articleId;

    // 自媒体或者爬虫
    public enum ArticleType{
        WEMEDIA,CRAWLER;
    }

}

一般在配置模块中新建kafka.messages包存放发送消息的实体类，消息内容的再次抽象，包含kafka变量。

package configuration.kafka.messages;

import com.lead.news.configuration.kafka.KafkaMessage;
import com.lead.news.model.mess.admin.SubmitArticleAuto;

/**
 * Title：文章提交的kafka消息审核实体类
 * Description：针对单通道和多通道进行区分
 * @author WZQ
 * @version 1.0.0
 * @date 2020/2/9
 */
public class SubmitArticleAuthMessage extends KafkaMessage<SubmitArticleAuto> {

    public SubmitArticleAuthMessage(){}

    public SubmitArticleAuthMessage(SubmitArticleAuto data){
        super(data);
    }

    // 消息区分，最好跟配置的名字一样
    @Override
    public String getType() {
        return "submit-article-auth";
    }
}

在工具类中的发送消息KafkaSender类加入新增的消息方法

    /**
     * 发送审核文章的消息
     * 消息名称，id，消息内容
     * @param message
     */
    public void sendSubmitArticleAuthMessage(SubmitArticleAuthMessage message) {
        this.sendMesssage(kafkaTopicConfig.getSubmitArticleAuth(), UUID.randomUUID().toString(), message);
    }
发送消息模块

配置类application.yml加入，对应配置模块的变量名，内容是这次模块的消息名称，被加载

# 文章提交的kafka消息审核名称
kafka:
  topic:
    submit-article-auth: wzq.topic.submit.article.auth.sigle.test

发送消息模块的kafka发送方法组件

package com.lead.news.media.kafka;

import com.lead.news.configuration.kafka.KafkaSender;
import com.lead.news.configuration.kafka.messages.SubmitArticleAuthMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * Title：提供者
 * Description：发送消息模块引用类
 * @author WZQ
 * @version 1.0.0
 * @date 2020/2/10
 */
@Component
public class AdminMessageSender {

    @Autowired
    KafkaSender kafkaSender;

    /**
     * 只发送审核文章行为消息
     * @param message
     */
    @Async
    public void sendMessage(SubmitArticleAuthMessage message){
        kafkaSender.sendSubmitArticleAuthMessage(message);
    }
}

加入service业务中

@Autowired
private AdminMessageSender adminMessageSender;

// 业务结束后发送消息
SubmitArticleAuthMessage message = new SubmitArticleAuthMessage();
SubmitArticleAuto submitArticleAuto = new SubmitArticleAuto();
// id带过去给消费者，这样就可以获取数据库数据进行操作
submitArticleAuto.setArticleId(wmNews.getId());
submitArticleAuto.setType(SubmitArticleAuto.ArticleType.WEMEDIA);
message.setData(submitArticleAuto);
adminMessageSender.sendMessage(message);
接收消息模块

appication.yml也得加入消息名称变量，内容保持一致

# 文章提交的kafka消息审核
kafka:
  topic:
    submit-article-auth: wzq.topic.submit.article.auth.sigle.test

定义接受消息后处理消息业务的监听器

package com.lead.news.admin.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.lead.news.admin.service.ReviewMediaArticleService;
import com.lead.news.configuration.kafka.KafkaListener;
import com.lead.news.configuration.kafka.KafkaTopicConfig;
import com.lead.news.configuration.kafka.messages.SubmitArticleAuthMessage;
import com.lead.news.model.mess.admin.SubmitArticleAuto;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * Title：消费者
 * Description：负责接收kafka消息
 * @author WZQ
 * @version 1.0.0
 * @date 2020/2/9
 */
@Component
@Log4j2
public class AutoReviewArticleListener implements KafkaListener<String,String> {

    @Autowired
    private KafkaTopicConfig kafkaTopicConfig;

    @Autowired
    private ObjectMapper mapper;

    // 处理消息业务的service，参数是发送消息模块发过来的，根据dto自己定义
    @Autowired
    private ReviewMediaArticleService reviewMediaArticleService;

    /*@Autowired
    private ReviewCrawlerArticleService reviewCrawlerArticleService;*/

    // 拿到消息名称
    @Override
    public String topic() {
        return kafkaTopicConfig.getSubmitArticleAuth();
    }

    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord, Consumer<?, ?> consumer) {
        String value = consumerRecord.value();
        log.info("接收到的消息为：{}"+value);
        try {
            // 拿到json数据
            SubmitArticleAuthMessage message = mapper.readValue(value, SubmitArticleAuthMessage.class);
            if(message!=null){
                SubmitArticleAuto.ArticleType type = message.getData().getType();
                if(type==SubmitArticleAuto.ArticleType.WEMEDIA){
                    Integer articleId = message.getData().getArticleId();
                    if(articleId!=null){
                        //审核文章信息
                        reviewMediaArticleService.autoReviewArticleByMedia(articleId);
                    }
                }/*else if(type==SubmitArticleAuto.ArticleType.CRAWLER){
                    Integer articleId = message.getData().getArticleId();
                    if(articleId!=null){
                        //审核爬虫文章信息
                        try {
                            reviewCrawlerArticleService.autoReivewArticleByCrawler(articleId);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }*/
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("处理自动审核文章错误:[{}],{}",value,e);
            throw new RuntimeException("WS消息处理错误",e);
        }
    }
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/149603.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【自学C++】C++整型

【自学C++】C++整型

C整型 C整型教程 C 的整型用来存放整数 类型 的数字&#xff0c;即不可以带小数&#xff0c;C 整型可以分为短整型 short&#xff0c;整型 int&#xff0c;长整型 long 和 long long 类型。 C整型取值范围 数据类型取值范围字节数short-32768 ~ 327672int-2147483648 ~ 214…
阅读更多...
小米万兆路由器里的Docker安装drawio

小米万兆路由器里的Docker安装drawio

小米2022年12月份发布了万兆路由器&#xff0c;里面可以使用Docker。 今天尝试在小米的万兆路由器里安装drawio 20.8.3。 准备工作 请参考https://engchina.blog.csdn.net/article/details/128515422的准备工作。 查看Docker Hub镜像信息 访问https://hub.docker.com/r/jgr…
阅读更多...
axios系列之并发

axios系列之并发

文章の目录一、axios.all(iterable)二、axios.spread(callback)写在最后处理并发请求的助手函数 一、axios.all(iterable) function a() {return axios.get("http://localhost/a"); } function b() {return axios.get("http://localhost/b"); }axios.all…
阅读更多...
【Linux 进程控制】进程创建、进程终止、进程等待、进程替换

【Linux 进程控制】进程创建、进程终止、进程等待、进程替换

1.进程创建&#xff08;fork&#xff09;#include<iostream> #include<unistd.h> using std::cout; using std::endl;int main() {if(fork()0){cout<<"child:"<<"I am child"<<endl;}else{cout<<"parent:"&…
阅读更多...
用555定时器接成的施密特触发器电路/滞回电压比较器中,用什么方法能调节回差电压的大小?包含工作原理与应用

用555定时器接成的施密特触发器电路/滞回电压比较器中,用什么方法能调节回差电压的大小?包含工作原理与应用

一、简答:电源电压或外接控制电压改变时&#xff0c;可以改变回差电压的大小。二、施密特触发器电路工作原理&#xff1a;滞回电压比较器&#xff0c;又名施密特触发器&#xff0c;有两个稳定状态&#xff0c;与一般触发器不同的是&#xff0c;施密特触发器采用电位触发方式&am…
阅读更多...
【免杀】通用shellcode原理及思路——FS段寄存器获取kernel32.dll基址逻辑、根据函数名进行查找逻辑、双指针循环遍历获取函数名称

【免杀】通用shellcode原理及思路——FS段寄存器获取kernel32.dll基址逻辑、根据函数名进行查找逻辑、双指针循环遍历获取函数名称

通用shellcode通用shellcode思路FS段寄存器获取kernel32.dll基址逻辑根据函数名进行查找逻辑双指针循环遍历获取函数字符串总结通用shellcode思路 1、保存相关字符串 user32.dll、LoadLibraryA、GetProcAddress、MessageBoxA、hello 51hook 2、通过fs寄存器获取kernel32.dll…
阅读更多...
8.Isaac教程--在Kaya上运行IsaacSDK

8.Isaac教程--在Kaya上运行IsaacSDK

在Kaya上运行IsaacSDK 构建自己的 NVIDIA Kaya 机器人后&#xff0c;请按照本页中的步骤在其上运行一些示例应用程序。 文章目录在Kaya上运行IsaacSDK安装和设置操纵杆应用程序跟随我的应用程序物体检测应用地图应用安装和设置 在 Kaya 上运行应用程序之前完成这些先决条件步…
阅读更多...
Jenkins+Git+Maven自动化部署配置

Jenkins+Git+Maven自动化部署配置

JenkinsGitMaven自动化部署配置基本思路1、jenkins安装maven依赖2、git安装3、Jenkins中新建任务3.1 git配置3.2 maven配置3.3 pom.xml配置3.4 build4、自动化发布到测试服务器并自动执行(Test-server)4.1 安装Publish Over SSH插件4.2 修改Post Steps配置4.3 执行构建5、小插曲…
阅读更多...
实战6 :基于mmdetection搭建Faster R-CNN实现酒液杂质检测

实战6 :基于mmdetection搭建Faster R-CNN实现酒液杂质检测

项目介绍： 本教程讲述如何解决依赖运动变化来区分目标的目标检测问题。以酒液杂质目标检测为例，讲述如何合理利用运动的酒液图像来设计算法进行杂质的检测。通过学习掌握依赖运动变化区分目标的检测任务的解决方法。首先我们对酒液杂质检验任务做了简要介绍。接下来再对数据…
阅读更多...
【C进阶】qsort函数详解

【C进阶】qsort函数详解

qsort函数前言qsort函数&#xff08;一&#xff09;引例&#xff1a;简单的冒泡排序&#xff08;二&#xff09;qsort函数接收1.介绍2.void*&#xff08;三&#xff09;使用1.用qsort实现一个比较整型的函数2.用qsort实现一个比较结构体的函数3.用qsort实现一个比较浮点型的函数…
阅读更多...
试读：目标检测定义及技术详解

试读：目标检测定义及技术详解

1.通用目标检测Generic Object Detection定义 目标检测旨在从图像、视频或者类似高维数据中定位大量预定义类别的物体实例&#xff0c;原始的图像、视频或类似数据经过数据预处理后&#xff0c;进入目标检测模型进行前向预测&#xff0c;最终得到数据中每个实例的位置以及该实…
阅读更多...
Compose 动画入门 (一) : animateXxxAsState 实现放大/缩小/渐变等效果

Compose 动画入门 (一) : animateXxxAsState 实现放大/缩小/渐变等效果

1. 前言 动画是Android Compose中很重要的一块内容。利用Compose可以简洁地定义动画&#xff0c;我们可以轻松快速地通过动画让应用变得生动有趣。 本文会介绍如何定义一个最简单的Compose动画&#xff0c;从而实现Compose动画。 1.1 定义一个Box 首先&#xff0c;我们先定义…
阅读更多...
关于volatile解决内存可见性问题（保证线程安全）

关于volatile解决内存可见性问题（保证线程安全）

Volatile是和内存可见性问题是密切相关的。先看下面一段代码&#xff0c;执行结果是什么&#xff1f; class MyCount{public int flag 0; } public class ThreadDemo15 {public static void main(String[] args) {MyCount myCount new MyCount();Thread t1 new Thread(()-&…
阅读更多...
B站涨粉十万+！B站up主如何吸引高质量粉丝？

B站涨粉十万+！B站up主如何吸引高质量粉丝？

如何在b站快速增长粉丝&#xff0c;b站如何快速涨粉丝&#xff0c;这是所有Up主都关心的问题&#xff0c;对于初来乍到B站的up主来说&#xff0c;发布的作品内容是极为重要的&#xff0c;B站最初的粉丝积累往往都是靠这些在B站生产的视频。作品是否精彩&#xff0c;能否吸引粉丝…
阅读更多...
AutoCAD打开文件提示“无法识别的版本，不能读取”

AutoCAD打开文件提示“无法识别的版本，不能读取”

今天遇到一个很奇怪的问题&#xff0c;在CAD中执行自己创建的命令并关闭文档之后&#xff0c;重新打开CAD提示“无法识别的版本&#xff0c;不能读取”错误对话框。 后来查询资料&#xff0c;发现这是CAD的一个老bug了。原因是该dwg文档所在的目录下某个文件的文件名以“无”开…
阅读更多...
MATLAB-常微分方程求解

MATLAB-常微分方程求解

MATLAB中可以用来求解常微分方程(组)的函数有ode23、 ode23s、 ode23t、 ode23tb 、ode45、ode15s和odel13等&#xff0c;见下表。它们的具体调用方法类似&#xff0c;为了方便后面的描述&#xff0c;在后面的介绍中将使用solver统一代替它们。函数的具体调用方法如下。[T,Y] s…
阅读更多...
简述 synchronized 和 ReentrantLock 之间的区别？

简述 synchronized 和 ReentrantLock 之间的区别？

相同点 synchronized 和 ReentrantLock 都是 Java 中提供的可重入锁。 可重入锁&#xff1a;什么是 “可重入”&#xff0c;可重入就是说某个线程已经获得某个锁&#xff0c;可以再次获取锁而不会出现死锁。 不同点 用法不同&#xff1a;synchronized 可以用来修饰普通方法、静…
阅读更多...
红中私教：使用wamp64配置靶场

红中私教：使用wamp64配置靶场

应朋友请求&#xff0c;出一篇配服务器的教程 首先安装软件 https://cowtransfer.com/s/9db1b9ad2c1d44 点击链接查看 [ wampserver3.3.0_x64.exe ] &#xff0c;或访问奶牛快传 cowtransfer.com 输入传输口令 pgs341 查看&#xff1b; 接下来&#xff0c;使用HbuilderX打开我…
阅读更多...
基于Paddle实现实例分割

基于Paddle实现实例分割

百度的Paddle这几年发展十分迅速&#xff0c;而且文档十分齐全&#xff0c;涉及到机器视觉的多个应用领域&#xff0c;感觉还是非常牛的&#xff0c;各种backbone&#xff0c;损失函数、数据增强手段以及NMS等&#xff0c;整体感觉复现的很全面&#xff0c;值得推荐学习。 本…
阅读更多...
搭建Redisson流程以及解读MutilLock源码解决分布式锁的主从一致性问题

搭建Redisson流程以及解读MutilLock源码解决分布式锁的主从一致性问题

搭建Redisson流程以及解读MutilLock源码解决分布式锁的主从一致性问题1、搭建3台独立主节点的redis服务2、创建java redisson客户端3、获取分布式锁4、分析获取锁源码getMultiLocktryLock(long waitTime, long leaseTime, TimeUnit unit)5、总结1、搭建3台独立主节点的redis服务…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023