Kafka系列(三）

本章内容继前两章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>