spring与kafka集成即spring-kafka实例讲解

发表于   |   分类于 左手代码 java   |   阅读(85)   |   评论(0)

一、消息系统模块说明

消息系统模块

  • common : 公共模块(一般放到maven私服中，供生产者和消费者加载)
  • producer :　生产者服务(一般单独部署，依赖common包)
  • consumer:　消费者服务(一般单独部署，依赖common包)

本文通过一个用户注册的案例来介绍spring与kafka客户端集成的技术实现

  • 用户注册：
    • 插入数据库等操作
    • 发送用户注册消息(生产者)，其他相关事务在用户注册后，发送邮件或其他的操作(消费者)

二、公共模块定义

1.公共模块主要内容
  • 消息内容实体(供生产者和消费者使用)
  • 消息内容序列化与反序列化工具类
    • 消息内容序列化：消息生产者将消息内容进行序列化发送到broker
    • 消息内容反序列化：消费者从broker拉取消息，将消息内容进行反序列化

注：消息为什么要序列化或反序列化呢？

因为我们发送的消息内容一般不可能是单纯的字符串，也是结构化数据，如一个数组列表，或者一个数据实体类，或者一个Map，因此，我们需要将结构化的数据转化为字符串，因此就要用到序列化类．

2.导入kafka依赖包
  1. <dependency>
  2.     <groupId>org.apache.kafka</groupId>
  3.     <artifactId>kafka-clients</artifactId>
  4.     <version>2.1.0</version>
  5. </dependency>
3.消息内容实体

根据业务需要，定义消息内容实体，一般可能与数据库表结构相关，这里以一个简单的用户注册的例子来说明，定义一个用户类

  1. @Data
  2. public class User implements Serializable {
  3.     private static final long serialVersionUID = -3247930189609406661L;
  5.     private long userId;
  6.     private String nickname;
  7.     private String email;
  8.     private String account;
  9.     private String password;
  10.     private int status;
  11.     private long createTime;
  13.     public User() {
  14.     }
  16.     public User(String nickname, String email, String account, String password, int status, long createTime) {
  17.         this.userId = userId;
  18.         this.nickname = nickname;
  19.         this.email = email;
  20.         this.account = account;
  21.         this.password = password;
  22.         this.status = status;
  23.         this.createTime = createTime;
  24.     }
  26.     public User(long userId, String nickname, String email, String account, String password, int status, long createTime) {
  27.         this.userId = userId;
  28.         this.nickname = nickname;
  29.         this.email = email;
  30.         this.account = account;
  31.         this.password = password;
  32.         this.status = status;
  33.         this.createTime = createTime;
  34.     }
  36.     @Override
  37.     public String toString() {
  38.         return "User{" +
  39.                 "userId=" + userId +
  40.                 ", nickname='" + nickname + '\'' +
  41.                 ", email='" + email + '\'' +
  42.                 ", account='" + account + '\'' +
  43.                 ", password='" + password + '\'' +
  44.                 ", status=" + status +
  45.                 ", createTime=" + createTime +
  46.                 '}';
  47.     }
  48. }

注：

  • 消息实体类一定要实现Serializable这个接口(因为消息内容要序列化，必须显示说明它是可序列化的)
4.定义消息内容序列化类和反序列化类
(1)定义一个转换工具类：用于字节数组与对象之间的转换
  1. import java.io.*;
  3. /**
  4.  * @author shixinke
  5.  * @version 1.0
  6.  * @Description
  7.  * @Date 19-2-1 下午3:02
  8.  */
  9. public class Converter {
  11.     /**
  12.      * bean转化为byte数组
  13.      * @param object
  14.      * @return
  15.      */
  16.     public static byte[] bean2ByteArray(Object object) {
  17.         byte[] bytes = null;
  18.         ByteArrayOutputStream byteArrayOutputStream = null;
  19.         ObjectOutputStream outputStream = null;
  20.         try {
  21.             /**
  22.              * 1.定义字节数组输出流和对象输出流
  23.              */
  24.             byteArrayOutputStream = new ByteArrayOutputStream();
  25.             outputStream = new ObjectOutputStream(byteArrayOutputStream);
  26.             /**
  27.              * 2.将对象写入输出流中
  28.              */
  29.             outputStream.writeObject(object);
  30.             outputStream.flush();
  31.             /**
  32.              * 3.字节数组输出流转化为字节数组
  33.              */
  34.             bytes = byteArrayOutputStream.toByteArray();
  35.         } catch (Exception ex) {
  36.             ex.printStackTrace();
  37.         } finally {
  38.             /**
  39.              * 4.关闭流,释放资源
  40.              */
  41.             if (outputStream != null) {
  42.                 try {
  43.                     outputStream.close();
  44.                 } catch (IOException e) {
  45.                     e.printStackTrace();
  46.                 }
  47.             }
  48.             if (byteArrayOutputStream != null) {
  49.                 try {
  50.                     byteArrayOutputStream.close();
  51.                 } catch (IOException e) {
  52.                     e.printStackTrace();
  53.                 }
  54.             }
  55.         }
  56.         return bytes;
  57.     }
  59.     /**
  60.      * byte数组转化为bean
  61.      * @param bytes
  62.      * @return
  63.      */
  64.     public static Object byteArray2Bean(byte[] bytes) {
  65.         Object object = null;
  66.         ByteArrayInputStream byteArrayInputStream = null;
  67.         ObjectInputStream inputStream = null;
  68.         try {
  69.             /**
  70.              * 1.定义字节数组输入流
  71.              */
  72.             byteArrayInputStream= new ByteArrayInputStream(bytes);
  73.             /**
  74.              * 2.定义输入流
  75.              */
  76.             inputStream = new ObjectInputStream(byteArrayInputStream);
  77.             /**
  78.              * 3.从输入流中读取对象数据
  79.              */
  80.             object = inputStream.readObject();
  81.         } catch (Exception ex) {
  82.             ex.printStackTrace();
  83.         } finally {
  84.             /**
  85.              * 4.关闭流,释放资源
  86.              */
  87.             if (inputStream != null) {
  88.                 try {
  89.                     inputStream.close();
  90.                 } catch (IOException e) {
  91.                     e.printStackTrace();
  92.                 }
  93.             }
  94.             if (byteArrayInputStream != null) {
  95.                 try {
  96.                     byteArrayInputStream.close();
  97.                 } catch (IOException e) {
  98.                     e.printStackTrace();
  99.                 }
  100.             }
  102.         }
  103.         return object;
  104.     }
  105. }
(2)定义序列化类
  2. import org.apache.kafka.common.serialization.Serializer;
  4. import java.util.Map;
  6. /**
  7.  * @author shixinke
  8.  * @version 1.0
  9.  * @Description
  10.  * @Date 19-2-1 下午2:56
  11.  */
  12. public class KafkaObjectSerializer implements Serializer<Object> {
  14.     @Override
  15.     public void configure(Map<String, ?> config, boolean isKey) {
  17.     }
  19.     /**
  20.      * 实现序列化的方法
  21.      * @param topic　主题
  22.      * @param data 消息内容数据
  23.      * @return
  24.      */
  25.     @Override
  26.     public byte[] serialize(String topic, Object data) {
  27.         return Converter.bean2ByteArray(data);
  28.     }
  30.     @Override
  31.     public void close() {
  33.     }
  34. }

注：

  • 序列化类主要是实现org.apache.kafka.common.serialization.Serializer这个类
    • serialize 方法：实现序列化的方法　
  • 序列化类主要是消息生产者使用，在发送消息前先消息内容进行序列化
(3)实现反序列化方法
  1. mport org.apache.kafka.common.serialization.Deserializer;
  3. import java.util.Map;
  5. /**
  6.  * @author shixinke
  7.  * @version 1.0
  8.  * @Description
  9.  * @Date 19-2-1 下午5:43
  10.  */
  11. public class KafkaObjectDeserializer implements Deserializer<Object> {
  12.     @Override
  13.     public void configure(Map<String, ?> config, boolean isKey) {
  15.     }
  17.     /**
  18.      * 反序列化方法
  19.      * @param topic　主题
  20.      * @param data　消息内容
  21.      * @return
  22.      */
  23.     @Override
  24.     public Object deserialize(String topic, byte[] data) {
  25.         return Converter.byteArray2Bean(data);
  26.     }
  28.     @Override
  29.     public void close() {
  31.     }
  32. }

注：

  • 反序列化类主要是实现org.apache.kafka.common.serialization.Deserializer这个类
    • deserialize 方法：实现反序列化的方法　
  • 反序列化类主要是消息消费者使用，在收到消息消费前，将生产者序列化的消息内容进行反序列化，还原成它本来的样子
  • 序列化和反序列化是一一对应的，消息结构一定要约定好，定义在一个公共模块中

三、消息生产者Producer

1.导入kafka依赖包

在项目maven配置pom.xml中加入：

  1. <!-- kafka包 -->
  2. <dependency>
  3.     <groupId>org.springframework.kafka</groupId>
  4.     <artifactId>spring-kafka</artifactId>
  5.     <version>2.2.3.RELEASE</version>
  6. </dependency>
  8. <!-- 自定义的公共模块　-->
  9. <dependency>
  10.     <groupId>com.shixinke.github.kafka.practise.common</groupId>
  11.     <artifactId>kafka-practise-common</artifactId>
  12.     <version>0.0.1-SNAPSHOT</version>
  13. </dependency>
2.kafka相关的应用配置
  1. spring.application.name=kafka-practise-producer
  2. spring.kafka.producer.batch-size=16384
  3. spring.kafka.producer.acks=all
  4. spring.kafka.producer.bootstrap-servers=localhost:9092
  5. spring.kafka.producer.buffer-memory=33554432
  6. spring.kafka.producer.client-id=${spring.application.name}
  7. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  8. spring.kafka.producer.value-serializer=com.shixinke.github.kafka.practise.common.util.KafkaObjectSerializer
  • spring.kafka.producer.batch-size : 批量发送消息的最大容量
  • spring.kafka.producer.acks　: 消息确认方式
  • spring.kafka.producer.bootstrap-servers : 连接的broker地址
  • spring.kafka.producer.buffer-memory : 发送缓冲区大小
  • spring.kafka.producer.client-id : 当前应用的标识
  • spring.kafka.producer.key-serializer : 消息key的序列化类
  • spring.kafka.producer.value-serializer : 消息内容的序列化类(指定到公共模块的序列化类)
3.编写kafka配置类
(1)配置参数类
  1. import lombok.Data;
  2. import org.springframework.boot.context.properties.ConfigurationProperties;
  3. import org.springframework.stereotype.Component;
  5. /**
  6.  * @author shixinke
  7.  * @version 1.0
  8.  * @Description
  9.  * @Date 19-2-2 下午4:33
  10.  */
  11. @Component
  12. @ConfigurationProperties(prefix = "spring.kafka.producer")
  13. @Data
  14. public class KafkaProducerProperties {
  15.     private String batchSize;
  16.     private String acks;
  17.     private String bootstrapServers;
  18.     private Long bufferMemory;
  19.     private String clientId;
  20.     private String keySerializer;
  21.     private String valueSerializer;
  22. }
  • 对应应用配置中的配置项
(2)kafka对象配置类
  2. import org.apache.kafka.clients.producer.ProducerConfig;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  6. import org.springframework.kafka.core.KafkaTemplate;
  8. import javax.annotation.Resource;
  9. import java.util.HashMap;
  10. import java.util.Map;
  12. /**
  13.  * @author shixinke
  14.  * @version 1.0
  15.  * @Description
  16.  * @Date 19-2-2 下午4:31
  17.  */
  18. @Configuration
  19. public class KafkaProducerConfiguration {
  21.     @Resource
  22.     private KafkaProducerProperties kafkaProducerProperties;
  24.     public Map<String, Object> producerConfig() {
  25.         Map<String, Object> props = new HashMap<String, Object>(8);
  26.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProducerProperties.getBootstrapServers());
  27.         props.put(ProducerConfig.ACKS_CONFIG, kafkaProducerProperties.getAcks());
  28.         props.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaProducerProperties.getBatchSize());
  29.         props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaProducerProperties.getBufferMemory());
  30.         props.put(ProducerConfig.CLIENT_ID_CONFIG, kafkaProducerProperties.getClientId());
  31.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaProducerProperties.getKeySerializer());
  32.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaProducerProperties.getValueSerializer());
  33.         return props;
  34.     }
  36.     @Bean
  37.     public <V> KafkaTemplate<String, V> kafkaTemplate() {
  38.         return new KafkaTemplate<String, V>(new DefaultKafkaProducerFactory<String, V>(producerConfig()));
  39.     }
  40. }
  • 通过kafka配置参数类来实例化KafkaTemplate类(调用生产者对象的类)
4.生产者服务类
(1)消息生产者服务接口
  1. import org.apache.kafka.clients.producer.RecordMetadata;
  3. import java.util.concurrent.Future;
  5. /**
  6.  * @author shixinke
  7.  * @version 1.0
  8.  * @Description
  9.  * @Date 19-2-1 下午2:05
  10.  */
  11. public interface KafkaMQProducerService {
  12.     public <K, V> Future<RecordMetadata> send(String topic, K key, V content);
  14. }
(2)消息生产者服务实现类
  1. import com.shixinke.github.kafka.practise.producer.service.KafkaMQProducerService;
  2. import org.apache.kafka.clients.producer.ProducerRecord;
  3. import org.apache.kafka.clients.producer.RecordMetadata;
  4. import org.springframework.kafka.core.KafkaTemplate;
  5. import org.springframework.stereotype.Service;
  7. import javax.annotation.Resource;
  8. import java.util.concurrent.Future;
  10. /**
  11.  * @author shixinke
  12.  * @version 1.0
  13.  * @Description
  14.  * @Date 19-2-1 下午2:06
  15.  */
  16. @Service
  17. public class KafkaMQProducerServiceImpl implements KafkaMQProducerService {
  19.     /**
  20.      * 在生产者配置中声明的KafkaTemplate的Bean类
  21.      */
  22.     @Resource
  23.     private KafkaTemplate kafkaProducer;
  25.     @Override
  26.     public <K, V> Future<RecordMetadata> send(String topic, K key, V content) {
  27.         ProducerRecord<String, V> record = new ProducerRecord<String, V>(topic, String.valueOf(key), content);
  28.         return kafkaProducer.send(record);
  29.     }
  31. }
5.业务服务(调用消息生产者的类)
(1)业务服务接口
  1. import com.shixinke.github.kafka.practise.common.data.Result;
  2. import com.shixinke.github.kafka.practise.producer.dto.UserDTO;
  4. public interface UserService {
  5.     public Result create(UserDTO userDTO);
  6. }
(2)业务服务实现
  1. mport com.shixinke.github.kafka.practise.common.data.Result;
  2. import com.shixinke.github.kafka.practise.common.data.User;
  3. import com.shixinke.github.kafka.practise.producer.dto.UserDTO;
  4. import com.shixinke.github.kafka.practise.producer.service.KafkaMQProducerService;
  5. import com.shixinke.github.kafka.practise.producer.service.UserService;
  6. import org.springframework.stereotype.Service;
  8. import javax.annotation.Resource;
  9. import java.time.Instant;
  11. /**
  12.  * @author shixinke
  13.  * @version 1.0
  14.  * @Description
  15.  * @Date 19-2-2 下午4:54
  16.  */
  17. @Service
  18. public class UserServiceImpl implements UserService {
  19.     @Resource
  20.     private KafkaMQProducerService producerService;
  22.     @Override
  23.     public Result create(UserDTO userDTO) {
  24.         User user = new User(userDTO.getNickname(), userDTO.getEmail(), userDTO.getAccount(), userDTO.getPassword(), 0, Instant.now().getEpochSecond());
  25.         /**
  26.          * 数据操作在此不再赘述(插入数据库，更新缓存)
  27.          */
  28.         user.setUserId(3000001);
  30.         /**
  31.          * 上面操作成功后，向消息队列发送一条消息，比如通知邮件服务要发邮件
  32.          */
  33.         producerService.send("user_reg", user.getUserId(), user);
  34.         return Result.success(user.getUserId());
  35.     }
  36. }
6.触发生产消息的事件
(1)接收用户端参数
  1. @Data
  2. public class UserDTO {
  3.     private String nickname;
  4.     private String account;
  5.     private String email;
  6.     private String password;
  8.     @Override
  9.     public String toString() {
  10.         return "UserDTO{" +
  11.                 "nickname='" + nickname + '\'' +
  12.                 ", account='" + account + '\'' +
  13.                 ", email='" + email + '\'' +
  14.                 ", password='" + password + '\'' +
  15.                 '}';
  16.     }
  17. }
(2)接收用户请求并处理
  1. @RestController
  2. @RequestMapping("/user")
  3. @Slf4j
  4. public class UserController {
  6.     @Resource
  7.     private UserService userService;
  9.     @PostMapping("register")
  10.     public Result register(UserDTO userDTO) {
  11.         Result result = null;
  12.         try {
  13.             result = userService.create(userDTO);
  14.         } catch (Exception ex) {
  15.             log.error("注册出错:", ex);
  16.             result = Result.error(ex.getMessage());
  17.         }
  18.         return result;
  19.     }
  20. }
(3)触发事件

在浏览器或其他工具中模拟用户注册：

在postman中模拟用户注册接口

四、消息消费者Consumer

1.导入kafka依赖包

在项目maven配置pom.xml中加入：

  1. <!-- kafka包 -->
  2. <dependency>
  3.     <groupId>org.springframework.kafka</groupId>
  4.     <artifactId>spring-kafka</artifactId>
  5.     <version>2.2.3.RELEASE</version>
  6. </dependency>
  8. <!-- 自定义的公共模块　-->
  9. <dependency>
  10.     <groupId>com.shixinke.github.kafka.practise.common</groupId>
  11.     <artifactId>kafka-practise-common</artifactId>
  12.     <version>0.0.1-SNAPSHOT</version>
  13. </dependency>
2.项目中kafka消费者相关配置
  1. spring.application.name=kafka-practise-consumer
  3. spring.kafka.consumer.bootstrap-servers=localhost:9092
  4. spring.kafka.consumer.group-id=user
  5. spring.kafka.consumer.client-id=${spring.application.name}
  6. spring.kafka.consumer.subscribe-topics=user_reg,user_update
  7. spring.kafka.consumer.enable-auto-commit=false
  8. spring.kafka.consumer.auto-commit-interval-ms=1000
  9. spring.kafka.consumer.concurrency=2
  10. spring.kafka.consumer.auto-offset-reset=earliest
  11. spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  12. spring.kafka.consumer.value-deserializer=com.shixinke.github.kafka.practise.common.util.KafkaObjectDeserializer
  • spring.kafka.consumer.bootstrap-servers : 连接的broker地址
  • spring.kafka.consumer.group-id : 消费者组名称
  • spring.kafka.consumer.client-id : 当前消费者标识
  • spring.kafka.consumer.subscribe-topics　: 消费的主题(自定义选项)
  • spring.kafka.consumer.enable-auto-commit : 是否自动提交
  • spring.kafka.consumer.auto-commit-interval-ms : 自动提交频率
  • spring.kafka.consumer.concurrency : 并发数
  • spring.kafka.consumer.auto-offset-reset : 当当前偏移量不存在或未指定偏移量时消费的偏移量
  • spring.kafka.consumer.key-deserializer：消息key的反序列化类
  • spring.kafka.consumer.value-deserializer:消息内容的反序列化类(使用公共模块中的反序列化类)
3.应用配置类
(1)消费者配置参数类
  1. import lombok.Data;
  2. import org.springframework.boot.context.properties.ConfigurationProperties;
  3. import org.springframework.stereotype.Component;
  5. @Component
  6. @ConfigurationProperties(prefix = "spring.kafka.consumer")
  7. @Data
  8. public class KafkaConsumerProperties {
  10.     private String bootstrapServers;
  11.     private String groupId;
  12.     private String clientId;
  13.     private String enableAutoCommit;
  14.     private String autoCommitIntervalMs;
  15.     private String autoOffsetReset;
  16.     private String keyDeserializer;
  17.     private String valueDeserializer;
  18.     private String subscribeTopics;
  19.     private int concurrency;
  20. }
  • 与应用配置文件中的kafka配置对应
(2)消费者对象配置类
  1. /**
  2.  * @author shixinke
  3.  * @version 1.0
  4.  * @Description
  5.  * @Date 19-2-1 下午4:05
  6.  */
  7. @Configuration
  8. public class KafkaConsumerConfiguration {
  10.     @Resource
  11.     private KafkaConsumerProperties kafkaProperties;
  13.     /**
  14.      * 通用的监听bean
  15.      * @return
  16.      */
  17.     @Bean
  18.     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
  19.         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  20.         factory.setConsumerFactory(consumerFactory());
  21.         factory.setConcurrency(kafkaProperties.getConcurrency());
  22.         factory.getContainerProperties().setPollTimeout(1500);
  23.         return factory;
  24.     }
  26.     /**
  27.      * 针对用户模块的监听bean
  28.      * @return
  29.      */
  30.     @Bean("userContainerFactory")
  31.     public ConcurrentKafkaListenerContainerFactory userContainerFactory() {
  32.         ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
  33.         factory.setConcurrency(kafkaProperties.getConcurrency());
  34.         factory.setBatchListener(true);
  35.         factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
  36.         factory.setConsumerFactory(consumerFactory());
  37.         return factory;
  38.     }
  40.     /**
  41.      * 消费者工厂
  42.      * @return
  43.      */
  44.     @Bean
  45.     public ConsumerFactory<String, String> consumerFactory() {
  46.         return new DefaultKafkaConsumerFactory<>(getProperties());
  47.     }
  50.     /**
  51.      * 配置参数设置(用于初始化)
  52.      * @return
  53.      */
  54.     public Map<String, Object> getProperties() {
  55.         Map<String, Object> props = new HashMap<String, Object>(8);
  56.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
  57.         props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getGroupId());
  58.         props.put(ConsumerConfig.CLIENT_ID_CONFIG, kafkaProperties.getClientId());
  59.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaProperties.getEnableAutoCommit());
  60.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProperties.getAutoOffsetReset());
  61.         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaProperties.getAutoCommitIntervalMs());
  62.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaProperties.getKeyDeserializer());
  63.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaProperties.getValueDeserializer());
  64.         return props;
  65.     }
  66. }
4.消息事件监听器(消费实现)
  1. @Component
  2. @Slf4j
  3. public class UserMessageListener {
  5.     @KafkaListener(id ="user_reg", topics = "user_reg", containerFactory = "userContainerFactory")
  6.     public void subscribeCreateUser(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
  7.         try {
  8.             log.info("收到消息：{}", records);
  9.             /**
  10.              * 手动提交偏移量
  11.              */
  12.             ack.acknowledge();
  13.         } catch (Exception ex) {
  14.             try {
  15.                 ack.wait();
  16.             } catch (InterruptedException e) {
  17.                 e.printStackTrace();
  18.             }
  19.         }
  20.     }
  21. }
  • 通过KafkaListener这个注解来标识这是一个消息事件监听
    • List> records : 消息内容列表
    • Acknowledgment ack : 消息确认对象
  1. 收到消息：[ConsumerRecord(topic = user_reg, partition = 0, offset = 1, CreateTime = 1549155259062, serialized key size = 7, serialized value size = 250, headers = RecordHeaders(headers = [], isReadOnly = false), key = 3000001, value = User(userId=3000001, nickname=诗心客, email=ishixinke@qq.com, account=shixinke, password=123456, status=0, createTime=1549155258))]

专题 更多...

分类 更多...

展开

标签 更多...

归档 更多...

展开