Kafka系列(三)

发布时间:2024年01月16日

本章内容继前两章1,2知识点后,开始结合java实际代码开发进行实践。其中三将讲述kafka座位消息中间件的代码样例,四将讲述kafka的stream流式计算处理。

消息中间件

首先说明一下,需要在虚拟机里面配置kafka,而kafka又依赖于zookeeper,它的一些元数据等都存储在zookeeper里。

如果需要了解如何安装kafka可以评论,这里不作为重点。

代码

pom依赖
      <!-- kafkfa -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
配置文件:

nacos里:

spring:
  kafka:
    bootstrap-servers: //自己的虚拟机地址
    producer:
      retries: 10
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: ${spring.application.name}
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

这里我给了一个通用的配置,具体的bootstrap里的配置我也给出一个示例

这里我将公有配置放到了common里面,大家可以配置属于自己的隔离级别,我这里也是为了可以复用方便。这里的配置需要在生产者端和消费者端都要配置。

spring:
  application:
    name: analysis-api #服务名称
  cloud:
    nacos:
      server-addr:  #nacos服务地址
      discovery: #服务注册的相关配置
        namespace: dev
        group: project
      config: #配置文件的相关信息
        namespace: dev #我的命名空间,无论是配置也好,服务也好,都在我的命名空间下
        group: project
        file-extension: yaml
        refresh-enabled: true
        extension-configs: #扩展配置,可以引用其他工程的配置文件
          - data-id: analysis-service-${spring.profiles.active}.yaml
            group: project
            refresh: true
        shared-configs:  #共享配置
          - data-id: swagger-${spring.profiles.active}.yaml
            group: common
            refresh: true
          - data-id: logging-${spring.profiles.active}.yaml
            group: common
            refresh: true
          - data-id: freemarker-config-dev.yaml
            group: common
            refresh: true
          - data-id: feign-${spring.profiles.active}.yaml
            group: common
            refresh: true
          - data-id: kafka-${spring.profiles.active}.yaml
            group: common
            refresh: true

  profiles:
    active: dev #环境名,拼接配置文件用的


在对应的service里注入

生产者:
   @Autowired
    KafkaTemplate kafkaTemplate;

    public void methodyourself(){
    //调用
    kafkaTemplate.send(Constants.TOPIC,JSON.toJSONString(所传的值));
}

这里为了方便管理,可以将主题定义成一个常量。

消费者
    @Autowired
    private WebService webService;

    @KafkaListener(topics = Constants.TOPIC)
    public void onMessage(String message){
        if(StringUtils.isNotBlank(message)){
            Map map = JSON.parseObject(message, Map.class);
            webService.updateByMap(map);
            log.info("文章修改,articleId={}",map.get("articleId"));
        }
    }

消费者监听相同的主题,然后生产者的时候对象是如何序列化的,消费者再给他按照对应的格式序列化回来,是字节序列号也好,json序列化也罢。

附上json依赖

     <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>

文章来源:https://blog.csdn.net/Java_Zzm/article/details/135632173
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。