spring cloud(boot)整合kafka

心已赠人 2023-02-16 09:27 18阅读 0赞

# 引入依赖 #

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

## 配置yml ##

spring:
      kafka:
        bootstrap-servers: 192.168.1.63:9092 #kafka地址
        template:
          default-topic: mykafka #topic 名
        listener:
          concurrency: 5 #指定listener 容器中的线程数，用于提高并发量
        consumer:
          group-id: myGroup # 指定默认消费者group id
          client-id: 200
          max-poll-records: 200
          auto-offset-reset: earliest # 最早未被消费的offset
        producer:
          batch-size: 1000 # 每次批量发送消息的数量
          retries: 3
          client-id: 200

### 构建生产者 ###

import org.apache.kafka.clients.producer.RecordMetadata;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.stereotype.Service;
    import org.springframework.util.concurrent.ListenableFuture;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    
    import java.util.concurrent.ExecutionException;
    
    /** * spring 2020/6/9 */
    @Service
    @Component
    public class Producer { 
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        /** * 发送消息到kafka */
        public RecordMetadata sendChannelMess(String topic, String message) { 
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
            RecordMetadata recordMetadata = null;
            try { 
                recordMetadata = future.get().getRecordMetadata();
            } catch (InterruptedException | ExecutionException e) { 
                e.printStackTrace();
                System.out.println("发送失败");
            }
            System.out.println("发送成功");
            System.out.println("partition:" + recordMetadata.partition());
            System.out.println("offset:" + recordMetadata.offset());
            System.out.println("topic:" + recordMetadata.topic());
    
            return recordMetadata;
        }
    
    }

#### 构建消费者 ####

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    import org.springframework.stereotype.Service;
    /** * spring 2020/6/9 */
    @Service
    @Component
    public class Consumer { 
        private static final Logger logger = LoggerFactory.getLogger(Consumer.class);
        /** * 有消息就读取,只读取消息value */
        @KafkaListener(topics = { "mykafka"})
        public void receiveMessage(String message){ 
             
            logger.error("***********************************************************");
            logger.error(message);
            logger.error("***********************************************************");
    
        }
    
    
    }