sring coud 2集成kafka

一时失言乱红尘 2021-09-28 02:42 226阅读 0赞

**安装zookeeper**

docker run --privileged=true --name zookeeper -p 2181:2181  -d zookeeper

**安装kafka**  
192.168.0.33为外网访问地址

docker run --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=192.168.0.33:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.0.33:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -d wurstmeister/kafka

**maven**

<dependency>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-starter-stream-kafka</artifactId>
    </dependency>

**application.yml**  
Spring Cloud 2 中 zk-nodes不用设置  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N1cGVybG92ZXJf_size_16_color_FFFFFF_t_70]

spring:
    	cloud:
        stream:
          bindings:
            shop_input:
              binder: kafka1
              consumer:
                headerMode: raw
              producer:
                headerMode: raw
              #绑定的kafka topic名称
              destination: shop-topic
              content-type: text/plain
            shop_output:
              binder: kafka1
              consumer:
                headerMode: raw
              producer:
                headerMode: raw
              destination: shop-topic
              content-type: text/plain
          binders:
          #可以配置多个kafka
            kafka1:
              type: kafka
              environment:
                spring:
                  cloud:
                    stream:
                      kafka:
                        binder:
                          #kafka地址
                          brokers: http://kafka:9092
                          auto-add-partitions: true
                          auto-create-topics: true
                          min-partition-count: 1

**Source**

import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    
    /**
     * @author yl
     */
    public interface MySource {
    
        String SHOP_OUTPUT = "shop_output";
    
        @Output(MySource.SHOP_OUTPUT)
        MessageChannel output();
    
    }

**Sink**

import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.messaging.SubscribableChannel;
    
    /**
     * @author yl
     */
    public interface MySink {
    
        String SHOP_INPUT = "shop_input";
     
        @Input(MySink.SHOP_INPUT)
        SubscribableChannel input();
    
    }

**KafkaSendTemplate**

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.messaging.support.MessageBuilder;
    
    /**
     * kafka消息发送模板
     *
     * @author yl
     */
    @EnableBinding(MySource.class)
    public class KafkaSendTemplate {
     
        @Autowired
        private MySource source;
     
        public void sendMessage(String msg) {
            try {
                source.output().send(MessageBuilder.withPayload(msg).build());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

**KafkaConsumer**

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    
    /**
     * kafka消息监听
     *
     * @author yl
     */
    @EnableBinding(MySink.class)
    @Slf4j
    public class KafkaConsumer {
        @Autowired
        SpCartService spCartService;
         
        @StreamListener(MySink.SHOP_INPUT)
        public void onReceive(String shopJson) {
            log.info(shopJson);
            ShopKafkaDTO shopKafkaDTO = JSONObject.parseObject(shopJson, ShopKafkaDTO.class);
            log.info("get Kafka message:{}", shopKafkaDTO);
    
        }
    }

**MyController**  
定义一个controller发送消息

@RestController
    public class MyController {
    	@Autowired
        private KafkaSendTemplate kafkaSendTemplate;
    
    	@GetMapping("/send")
        public void sendMessage(@RequestParam("message") String message) {
            kafkaSendTemplate.sendMessage(message);
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N1cGVybG92ZXJf_size_16_color_FFFFFF_t_70]: /images/20210923/1b5b8af39087450184e2b9199ccc9e47.png