spring-kafka多线程顺序消费

业务场景

创新互联是一家专注于成都网站制作、做网站与策划设计,宾川网站建设哪家好?创新互联做网站,专注于网站建设10年,网设计领域的专业建站公司;建站业务涵盖:宾川等地区。宾川做网站价格咨询:028-86922220

我们公司是做共享充电宝的业务的。有一些比较大的代理商或者ka商户，他们需要了解到他们自己下面的商户的订单数据，这些订单数据需要由我们推送给他们。

大致架构为数据部门通过canal订阅订单表的数据，然后推送到kafka ，我们订阅数据部门kafka获取到代理商下商户的实时订单数据再推送给代理商。比如，代理商下商户产生了一笔订单，整个过程会产生，订单生成，订单已支付，充电宝已被取走，充电宝已归还等多种状态的订单消息，我们需要实时把这些订单消息推送给代理商。我们的业务场景需要消息的顺序推送和多线程并发消费以提高性能

kafka多线程消费方案

消费者程序启动多个线程，每个线程维护专属的KafkaConsumer实例，负责完整的消息获取、消息处理

流程。如下图所示:

消费者程序使用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑。获取消息的线程可以 是一个，也可以是多个，每个线程维护专属的KafkaConsumer实例，处理消息则交由特定的线程池来 做，从而实现消息获取与消息处理的真正解耦。具体架构如下图所示:

这两种方案孰优孰劣呢?应该说是各有千秋。这两种方案的优缺点，我们先来看看下面这张表格。

kafka怎么保证顺序消费

保证顺序消费，需要满足如下条件

保证相同订单编号的消息需要发送到同一个分区。

@Configuration

public class SenderConfig {

@Value("${kafka.bootstrap-servers}")

private String bootstrapServers;

@Bean

public Map producerConfigs() {

Map props = new HashMap<>();

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

return props;

}

@Bean

public ProducerFactory producerFactory() {

return new DefaultKafkaProducerFactory<>(producerConfigs());

}

@Bean

public KafkaTemplate kafkaTemplate() {

return new KafkaTemplate<>(producerFactory());

}

@Bean

public Sender sender() {

return new Sender();

}

}

public class Sender {

@Autowired

private KafkaTemplate kafkaTemplate;

public void send(String topic, String data) {

kafkaTemplate.send(topic, data);

}

public void send(String topic, int partition, String data) {

kafkaTemplate.send(topic, partition, data);

}

}

@RunWith(SpringRunner.class)

@SpringBootTest

public class SpringKafkaApplicationTest {

private static String BATCH_TOPIC = "batch.t";

private static Integer PARTITIONS = 6;

/**

* 已支付

*/

private static Integer PAYED_STATUS = 2;

/**

* 已取走

*/

private static Integer SEND_BACK_STATUS = 3;

@Autowired

private Sender sender;

private static DelayQueue delayQueue = new DelayQueue();

@Test

public void testReceive() throws Exception {

for (int i = 1; i < 50; i++) {

Integer orderNum = 800010 + i;

Integer orderPrice = RandomUtil.randomInt(1, 20);

// 用户支付成功，订单状态为支付成功

OrderDTO order = new OrderDTO(orderNum, orderPrice, PAYED_STATUS);

// 发送支付成功订单消息到对应的kafka分区

Integer destinationPartition = orderNum % PARTITIONS;

sender.send(BATCH_TOPIC, destinationPartition, JSONUtil.toJsonStr(order));

// 创建任务放入延迟队列(模拟用户支付成功到取走充电宝花费的时间)

long delayTime = 200;

OrderTask orderTask = new OrderTask(delayTime, order);

delayQueue.offer(orderTask);

}

while (true) {

// 用户取走充电宝,订单状态更改为 已取走

OrderTask orderTask = (OrderTask) delayQueue.take();

OrderDTO orderDTO = orderTask.getOrderDTO();

Integer destinationPartition = orderDTO.getOrderNum() % PARTITIONS;

orderDTO.setOrderStatus(SEND_BACK_STATUS);

// 发送已取走订单消息到对应的kafka 分区

sender.send(BATCH_TOPIC, destinationPartition, JSONUtil.toJsonStr(orderDTO));

}

}

}

可以看出我们通过订单号对分区数进行取余，来确定该消息发送到哪一个分区，保证相同订单号的消息被发送到相同的分区。当然也可以对字符串这些进行hash ，获得hash值来对分区数取余

Integer destinationPartition=orderDTO.getOrderNum()%PARTITIONS;

保证同一个分区的消息由同一个线程来消费。

我们的业务场景需要采用多线程方案一来处理我们的业务

普通方式实现方案一

public class KafkaConsumerRunner implements Runnable {

private final AtomicBoolean closed = new AtomicBoolean(false);

private final KafkaConsumer consumer;

public KafkaConsumerRunner(KafkaConsumer consumer) {

this.consumer = consumer;

}

@Override

public void run() {

try {

consumer.subscribe(Arrays.asList("topic"));

while (!closed.get()) {

// 执行消息处理逻辑

ConsumerRecords records = consumer.poll(10000);

}

} catch (Exception e) {

// Ignore exception if closing

if (!closed.get()) {

throw e;

}

} finally {

consumer.close();

}

}

/**

* Shutdown hook which can be called from a separate thread

*/

public void shutdown() {

closed.set(true);

consumer.wakeup();

}

}

spring-kafka为我们做的封装

消费者相关配置：

这里我们需要注意的是factory.setConcurrency(4)。

这个是配置主要是设置KafkaConsumer的数量，最大为topic 的分区数。当然你如果设置的值超过topic 分区数，spring-kafka 还是只会为我们创建最大分区数的KafkaConsumer数量，也就是创建KafkaConsumer数量能少于分区数，但不会超过分区数。少于分区数的话，一个KafkaConsumer会消费多个分区的数据，保证所有的分区数据都有对应的KafkaConsumer来进行消费;但不会出现多个KafkaConsumer消费同一个分区的情况，因为如果是这样也就无法保证消息的顺序消费机制。

一般情况下如果数据量较大，我们需要把此值设置为topic分区数，这样一个KafkaConsumer消费一个分区的数据，提高数据的并发消费能力。

@Configuration

@EnableKafka

public class ReceiverConfig {

@Value("${kafka.bootstrap-servers}")

private String bootstrapServers;

@Bean

public Map consumerConfigs() {

Map props = new HashMap<>();

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

props.put(ConsumerConfig.GROUP_ID_CONFIG, "batch");

// maximum records per poll

props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100");

return props;

}

@Bean

public ConsumerFactory consumerFactory() {

return new DefaultKafkaConsumerFactory<>(consumerConfigs());

}

@Bean(name = "kafkaListenerContainerFactory")

public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {

ConcurrentKafkaListenerContainerFactory factory =

new ConcurrentKafkaListenerContainerFactory<>();

factory.setConsumerFactory(consumerFactory());

// enable batch listening

factory.setBatchListener(true);

factory.setConcurrency(4);

return factory;

}

@Bean

public Receiver receiver() {

return new Receiver();

}

}

Receiver 代码

public class Receiver {

@Autowired

private PushOrderService pushOrderService;

private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);

private static final String BATCH_TOPIC = "batch.t";

@KafkaListener(topics = BATCH_TOPIC, containerFactory = "kafkaListenerContainerFactory")

public void receivePartitions(List data,

@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List partitions,

@Header(KafkaHeaders.OFFSET) List offsets) {

for (int i = 0; i < data.size(); i++) {

Long threadId = Thread.currentThread().getId();

// 向第三方推送订单消息

String orderStr = data.get(i);

pushOrderService.pushOrderToPlatform(orderStr);

OrderDTO orderDTO = JSONUtil.toBean(orderStr, OrderDTO.class);

LOGGER.info("推送订单消息成功,订单号为:{},状态为:{},分区为{},处理线程为:{}", orderDTO.getOrderNum(), orderDTO.getOrderStatus(), partitions.get(i), threadId);

}

}

}

/**

* 模拟网络推送订单信息给第三方平台

*/

@Service

public class PushOrderService {

/**

* 已支付

*/

private static Integer PAYED_STATUS = 2;

public void pushOrderToPlatform(String orderString) {

// 模拟网络推送订单信息给第三方平台(同步推送)

OrderDTO orderDTO = JSONUtil.toBean(orderString, OrderDTO.class);

// 已支付 订单消息

if (orderDTO.getOrderStatus().equals(PAYED_STATUS)) {

ThreadUtil.sleep(500);

} else {

// 已取走 订单消息

ThreadUtil.sleep(200);

}

}

}

测试结果： 无锡做人流手术多少钱 http://www.ytsg029.com/

16:17:47.026 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800014,状态为:2,分区为4,处理线程为:67

16:17:47.026 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800012,状态为:2,分区为2,处理线程为:66

16:17:47.534 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800020,状态为:2,分区为4,处理线程为:67

16:17:47.534 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800018,状态为:2,分区为2,处理线程为:66

16:17:48.035 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800026,状态为:2,分区为4,处理线程为:67

16:17:48.036 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800024,状态为:2,分区为2,处理线程为:66

16:17:48.537 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800015,状态为:2,分区为5,处理线程为:67

16:17:48.539 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800016,状态为:2,分区为0,处理线程为:66

16:17:49.044 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800022,状态为:2,分区为0,处理线程为:66

16:17:49.045 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800021,状态为:2,分区为5,处理线程为:67

16:17:49.546 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800013,状态为:2,分区为3,处理线程为:67

16:17:49.547 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800028,状态为:2,分区为0,处理线程为:66

16:17:50.051 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800019,状态为:2,分区为3,处理线程为:67

16:17:50.051 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800011,状态为:2,分区为1,处理线程为:66

16:17:50.554 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800025,状态为:2,分区为3,处理线程为:67

16:17:50.554 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800017,状态为:2,分区为1,处理线程为:66

16:17:51.060 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800023,状态为:2,分区为1,处理线程为:66

16:17:51.576 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800034,状态为:2,分区为0,处理线程为:66

16:17:51.579 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800031,状态为:2,分区为3,处理线程为:67

16:17:51.583 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800032,状态为:2,分区为4,处理线程为:70

16:17:51.583 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800027,状态为:2,分区为5,处理线程为:72

16:17:52.079 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800040,状态为:2,分区为0,处理线程为:66

16:17:52.083 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800037,状态为:2,分区为3,处理线程为:67

16:17:52.088 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800038,状态为:2,分区为4,处理线程为:70

16:17:52.088 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800033,状态为:2,分区为5,处理线程为:72

16:17:52.583 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800046,状态为:2,分区为0,处理线程为:66

16:17:52.588 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800043,状态为:2,分区为3,处理线程为:67

16:17:52.589 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800044,状态为:2,分区为4,处理线程为:70

16:17:52.590 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800039,状态为:2,分区为5,处理线程为:72

16:17:53.089 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800052,状态为:2,分区为0,处理线程为:66

16:17:53.091 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800050,状态为:2,分区为4,处理线程为:70

16:17:53.091 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800049,状态为:2,分区为3,处理线程为:67

16:17:53.095 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800045,状态为:2,分区为5,处理线程为:72

16:17:53.591 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800058,状态为:2,分区为0,处理线程为:66

16:17:53.592 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800056,状态为:2,分区为4,处理线程为:70

16:17:53.593 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800055,状态为:2,分区为3,处理线程为:67

16:17:53.600 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800051,状态为:2,分区为5,处理线程为:72

16:17:53.795 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800016,状态为:3,分区为0,处理线程为:66

16:17:53.796 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800013,状态为:3,分区为3,处理线程为:67

16:17:53.796 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800014,状态为:3,分区为4,处理线程为:70

16:17:54.000 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800020,状态为:3,分区为4,处理线程为:70

16:17:54.000 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800019,状态为:3,分区为3,处理线程为:67

16:17:54.000 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800022,状态为:3,分区为0,处理线程为:66

16:17:54.101 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800057,状态为:2,分区为5,处理线程为:72

16:17:54.205 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800026,状态为:3,分区为4,处理线程为:70

16:17:54.206 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800025,状态为:3,分区为3,处理线程为:67

16:17:54.206 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800028,状态为:3,分区为0,处理线程为:66

16:17:54.306 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800015,状态为:3,分区为5,处理线程为:72

16:17:54.410 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800037,状态为:3,分区为3,处理线程为:67

16:17:54.410 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800034,状态为:3,分区为0,处理线程为:66

16:17:54.410 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800032,状态为:3,分区为4,处理线程为:70

16:17:54.510 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800021,状态为:3,分区为5,处理线程为:72

16:17:54.614 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800031,状态为:3,分区为3,处理线程为:67

16:17:54.615 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800046,状态为:3,分区为0,处理线程为:66

16:17:54.615 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800038,状态为:3,分区为4,处理线程为:70

16:17:54.711 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800027,状态为:3,分区为5,处理线程为:72

16:17:54.820 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800043,状态为:3,分区为3,处理线程为:67

16:17:54.820 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800040,状态为:3,分区为0,处理线程为:66

16:17:54.820 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800056,状态为:3,分区为4,处理线程为:70

16:17:54.914 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800033,状态为:3,分区为5,处理线程为:72

16:17:55.025 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800044,状态为:3,分区为4,处理线程为:70

16:17:55.025 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800052,状态为:3,分区为0,处理线程为:66

16:17:55.025 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800055,状态为:3,分区为3,处理线程为:67

16:17:55.118 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800039,状态为:3,分区为5,处理线程为:72

16:17:55.231 [org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800050,状态为:3,分区为4,处理线程为:70

16:17:55.231 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800049,状态为:3,分区为3,处理线程为:67

16:17:55.231 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800058,状态为:3,分区为0,处理线程为:66

16:17:55.321 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800057,状态为:3,分区为5,处理线程为:72

16:17:55.525 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800051,状态为:3,分区为5,处理线程为:72

16:17:55.728 [org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800045,状态为:3,分区为5,处理线程为:72

16:17:55.735 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800029,状态为:2,分区为1,处理线程为:66

16:17:55.737 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800030,状态为:2,分区为2,处理线程为:67

16:17:56.239 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800036,状态为:2,分区为2,处理线程为:67

16:17:56.239 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800035,状态为:2,分区为1,处理线程为:66

16:17:56.743 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800042,状态为:2,分区为2,处理线程为:67

16:17:56.743 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800041,状态为:2,分区为1,处理线程为:66

16:17:57.247 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800047,状态为:2,分区为1,处理线程为:66

16:17:57.247 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800048,状态为:2,分区为2,处理线程为:67

16:17:57.751 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800053,状态为:2,分区为1,处理线程为:66

16:17:57.751 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800054,状态为:2,分区为2,处理线程为:67

16:17:57.953 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800012,状态为:3,分区为2,处理线程为:67

16:17:58.159 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800018,状态为:3,分区为2,处理线程为:67

16:17:58.256 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800059,状态为:2,分区为1,处理线程为:66

16:17:58.361 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800024,状态为:3,分区为2,处理线程为:67

16:17:58.457 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800011,状态为:3,分区为1,处理线程为:66

16:17:58.566 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800048,状态为:3,分区为2,处理线程为:67

16:17:58.662 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800017,状态为:3,分区为1,处理线程为:66

16:17:58.771 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800042,状态为:3,分区为2,处理线程为:67

16:17:58.868 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800023,状态为:3,分区为1,处理线程为:66

16:17:58.975 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800030,状态为:3,分区为2,处理线程为:67

16:17:59.073 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800029,状态为:3,分区为1,处理线程为:66

16:17:59.177 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800036,状态为:3,分区为2,处理线程为:67

16:17:59.279 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800041,状态为:3,分区为1,处理线程为:66

16:17:59.383 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800054,状态为:3,分区为2,处理线程为:67

16:17:59.481 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800035,状态为:3,分区为1,处理线程为:66

16:17:59.685 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800053,状态为:3,分区为1,处理线程为:66

16:17:59.891 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800047,状态为:3,分区为1,处理线程为:66

16:18:00.092 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - 推送订单消息成功,订单号为:800059,状态为:3,分区为1,处理线程为:66

完整代码