kafka集成（二）springboot整合 kafka

绝地灬酷狼 2022-04-24 06:14 271阅读 0赞

实例一、发送普通字符串消息：

1、pom：

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

2、properties配置文件：

#kafka
    # 指定kafka 代理地址，可以多个
    spring.kafka.bootstrap-servers=localhost:9092
    # 指定默认消费者group id
    spring.kafka.consumer.group-id=myGroup
    # 指定默认topic id
    spring.kafka.template.default-topic= my-replicated-topic
    # 指定listener 容器中的线程数，用于提高并发量
    spring.kafka.listener.concurrency= 3
    # 每次批量发送消息的数量
    spring.kafka.producer.batch-size= 1000
    #key-value序列化反序列化
    #spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    #spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    #spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    #spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.buffer-memory=524288

3、时间Util:

public class DateUtil {
    
    	public static String getCurentTime(){
    		Date date = new Date();
    		SimpleDateFormat dateFormat =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    		return dateFormat.format(date);
    	}
    }

4、生产者：

package com.wtyy.rest;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Service;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.wtyy.util.DateUtil;
    
    /**
     * @Description: 生产者
     */
    @Service
    public class KafkaProducer {
    	@Autowired
    	private KafkaTemplate kafkaTemplate;
     
    	/**
    	 * 发送消息到kafka,主题为test
    	 */
    	public void sendTest(){
            kafkaTemplate.send("test","hello,kafka  "  + DateUtil.getCurentTime());
        }
    }

5、消费者，监听生产者的消息：

package com.wtyy.rest;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Service;
    
    @Service
    public class KafkaConsumer {
    	protected static Logger logger=LoggerFactory.getLogger(KafkaConsumer.class); 
    	/**
         * 监听test主题,有消息就读取
         * @param message
         */
        @KafkaListener(topics = {"test"})
        public void consumer(String message){
        	System.err.println("test topic message : "+ message);
        }
    }

6、定时发送消息（定时调用生产者）：

package com.wtyy.rest;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Service;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    @Service
    public class KafkaScheduled {
    	private static Logger logger = LoggerFactory.getLogger(KafkaScheduled.class);
    	@Autowired
        private KafkaProducer kafkaSender;
    	
    	// 然后每隔20ms执行一次
    	@Scheduled(fixedRate = 1000 * 20)
    	public void testKafka() throws Exception {
    		logger.info("KafkaScheduled...start");
    		kafkaSender.sendTest();
    	}
    }

7、启动类：

package com.wtyy;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.scheduling.annotation.EnableScheduling;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    /** 
     * @Description: 启动入口   
     */
    @SpringBootApplication
    @EnableScheduling
    public class KafkaApplication {
     
    	protected static Logger logger=LoggerFactory.getLogger(KafkaApplication.class); 
     
    	public static void main(String[] args) {
    		SpringApplication.run(KafkaApplication.class, args);
    		logger.info("----------------SpringBoot Start Success-------------------");
    	}
    }

启动后可以看到消费者不停的打印消息：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70][]

实例二：接着上面的例子，发送对象，一般转成json字符串：

1、实体类：

package com.wtyy.dto;
    
    import java.io.Serializable;
    
    
    public class Message implements Serializable{
    	
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = -8921653281090412125L;
    
    	private String msgId;
    	
    	private String content;
    	
    
    	/**这里省略get和set方法
        */
    
    	@Override
    	public String toString() {
    		return "Message [msgId=" + msgId + ", content=" + content + "]";
    	}
    
    	
    }

2、生产者：

package com.wtyy.rest;
    
    import java.util.UUID;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Service;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.alibaba.fastjson.JSON;
    import com.wtyy.dto.Message;
    import com.wtyy.util.DateUtil;
    
    /**
     * @Description: 生产者
     */
    @Service
    public class KafkaProducer {
    	@Autowired
    	private KafkaTemplate kafkaTemplate;
     
    	/**
    	 * 发送消息到kafka,主题为test
    	 */
    	public void sendTest(){
            kafkaTemplate.send("test","hello,kafka  "  + DateUtil.getCurentTime());
        }
    	
    
         /**
    	 * 发送消息到kafka,主题为msg
    	 */
    	public void sendMessageDTO(){
    		String uuid = UUID.randomUUID().toString();
    		Message message = new Message();
    		message.setMsgId(uuid);
    		message.setContent(DateUtil.getCurentTime()+"时产生消息"+uuid);
    		kafkaTemplate.send("msg",JSON.toJSONString(message));
    	}
    }

3、消费者：

package com.wtyy.rest;
    
    import java.util.List;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Service;
    
    import com.alibaba.fastjson.JSON;
    import com.wtyy.dto.Message;
    
    @Service
    public class KafkaConsumer {
    	protected static Logger logger=LoggerFactory.getLogger(KafkaConsumer.class); 
    	/**
         * 监听test主题,有消息就读取
         * @param message
         */
        @KafkaListener(topics = {"test"})
        public void consumer(String message){
        	System.out.println("test topic message : "+ message);
        }
        
        /**
         * 监听msg主题消息
         */
        @KafkaListener(topics = {"msg"})
        public void consumeMsgDTO(String message){
        	Message messageDTO = JSON.parseObject(message, Message.class);
        	System.err.println("msg接收到消息:"+messageDTO);
        }
    }

4、定时器：

package com.wtyy.rest;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Service;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    @Service
    public class KafkaScheduled {
    	private static Logger logger = LoggerFactory.getLogger(KafkaScheduled.class);
    	@Autowired
        private KafkaProducer kafkaSender;
    	
    	// 每隔20ms执行一次
    	@Scheduled(fixedRate = 1000 * 20)
    	public void testKafka() throws Exception {
    		logger.info("KafkaScheduled...start");
    		kafkaSender.sendTest();
    	}
    	
    	
    	/**
    	 * 一分钟执行一次，发送一次消息
    	 */
    	@Scheduled(cron = "0 0/1 * * * ? ")
    	public void msgKafka() throws Exception {
    		logger.info("KafkaMsgScheduled...start");
    		kafkaSender.sendMessageDTO();
    	}
    }

启动后：后台打印：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 1][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70]: /images/20220218/6f911cffa1394efa87134d9e3ca00bbe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 1]: /images/20220218/46bf70c7b09f40b89a85b3a78e7a3887.png