释放双眼，带上耳机，听听看~！

Spring Boot 2.0.4 集成 Kafka 2.0.0。
项目源码地址：https://gitee.com/lilyssh/high-concurrency

一、简介

kafka是一种高吞吐量的分布式发布订阅消息系统。kafka对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。无论是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。
Kafka的安装请参考文章：Kafka的安装与使用。

二、使用方法

（1）添加依赖


1
2
3
4
5
6
7
1&lt;dependency&gt;

2    &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;

3    &lt;artifactId&gt;spring-kafka&lt;/artifactId&gt;

4    &lt;version&gt;2.1.10.RELEASE&lt;/version&gt;

5&lt;/dependency&gt;

6

7

（2）在application.yml中添加配置

官方文档说只要配置两个必要项就可以了，spring.kafka.consumer.group-id和spring.kafka.consumer.auto-offset-reset。此处对其他配置稍作解释。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
1spring:

2  kafka:

3    # 指定kafka代理地址，brokers集群。

4    bootstrap-servers: ssh.qianxunclub.com:9092

5    producer:

6      # 发送失败重试次数。

7      retries: 0

8      # 每次批量发送消息的数量 批处理条数：当多个记录被发送到同一个分区时，生产者会尝试将记录合并到更少的请求中。这有助于客户端和服务器的性能。

9      batch-size: 16384

10      # 32MB的批处理缓冲区。

11      buffer-memory: 33554432

12      # 指定消息key和消息体的编解码方式。

13      key-serializer: org.apache.kafka.common.serialization.StringSerializer

14      value-serializer: org.apache.kafka.common.serialization.StringSerializer

15    consumer:

16      # 消费者群组ID，发布-订阅模式，即如果一个生产者，多个消费者都要消费，那么需要定义自己的群组，同一群组内的消费者只有一个能消费到消息。

17      group-id: kafka_order_group

18      auto-offset-reset: earliest

19      # 如果为true，消费者的偏移量将在后台定期提交。

20      enable-auto-commit: true

21      # 自动提交周期

22      auto-commit-interval: 100

23      # 指定消息key和消息体的编解码方式。

24      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

25      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

26

27

（3）消息发送类


1
2
3
1package cn.lilyssh.common.kafka.provider;

2

3

import com
.google
.gson
.Gson
;

import com
.google
.gson
.GsonBuilder
;

import lombok
.extern
.slf4j
.Slf4j
;

import org
.springframework
.beans
.factory
.annotation
.Autowired
;

import org
.springframework
.stereotype
.Component
;

import org
.springframework
.kafka
.core
.KafkaTemplate
;

@Component

@Slf4j

public
class
KafkaSender
{

@Autowired

private KafkaTemplate
<String
,String

kafkaTemplate

;

private Gson gson

new
GsonBuilder
(
)
.
create
(
)
;

//发送消息方法

public
void
send
(String topic
,String key
,Object message
)
{
kafkaTemplate
.
send
(topic
,key
,gson
.
toJson
(message
)
)
;
log
.
info
(
"+++++++++++++++++++++ message = {}"
, gson
.
toJson
(message
)
)
;

}

此处关键代码为kafkaTemplate.send()，参数topic是Kafka里的topic，这个topic在 Java程序中是不需要提前在Kafka中设置的，因为它会在发送的时候自动创建你设置的topic， gson.toJson(message)是消息内容。

（4）在下单业务中调用消息发送


1
2
3
1package cn.lilyssh.order.provider.service;

2

3

import cn
.lilyssh
.common
.kafka
.provider
.KafkaSender
;

import cn
.lilyssh
.order
.api
.model
.request
.OrderInsertReq
;

import cn
.lilyssh
.order
.api
.service
.OrderServiceApi
;

import cn
.lilyssh
.order
.provider
.dao
.entity
.OrderEntity
;

import com
.alibaba
.dubbo
.config
.annotation
.Service
;

import lombok
.AllArgsConstructor
;

import lombok
.extern
.slf4j
.Slf4j
;

import org
.springframework
.beans
.BeanUtils
;

import org
.springframework
.stereotype
.Component
;

import org
.springframework
.util
.StringUtils
;

import java
.math
.BigDecimal
;

import java
.util
.*
;

@Slf4j

@Service

@Component

@AllArgsConstructor

public
class
OrderService
implements
OrderServiceApi
{

private KafkaSender kafkaSender
;

/**
* 保存到kafka
* @param orderInsertReq
* @return
*/

@Override

public
void
saveByKafka
(OrderInsertReq orderInsertReq
)
{
OrderEntity orderEntity

new
OrderEntity
(
)
;

//直接写入数据库太慢，引起dubbo超时，导致调用多次，此处需要改造成kafka异步写入。
BeanUtils
.
copyProperties
(orderInsertReq
,orderEntity
)
;
kafkaSender
.
send
(
“placeOrder”
, orderEntity
.
getUserId
(
)
.
toString
(
)
, orderEntity
)
;

}

（5）消息接收类


1
2
3
1package cn.lilyssh.order.provider.kafka.consumer;

2

3

import cn
.lilyssh
.order
.provider
.dao
.entity
.OrderEntity
;

import cn
.lilyssh
.order
.provider
.dao
.repository
.OrderRepository
;

import com
.google
.gson
.Gson
;

import lombok
.AllArgsConstructor
;

import lombok
.extern
.slf4j
.Slf4j
;

import org
.apache
.kafka
.clients
.consumer
.ConsumerRecord
;

import org
.springframework
.kafka
.annotation
.KafkaListener
;

import org
.springframework
.stereotype
.Component
;

import java
.util
.Optional
;

@Component

@Slf4j

@AllArgsConstructor

public
class
KafkaReceiver
{


1
2
3
4
5
6
7
8
9
10
11
12
13
1&lt;span class=&quot;token keyword&quot;&gt;private&lt;/span&gt; OrderRepository orderRepository&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

2&lt;span class=&quot;token comment&quot;&gt;/**

3 * 监听下单

4 */&lt;/span&gt;

5&lt;span class=&quot;token annotation punctuation&quot;&gt;@KafkaListener&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;topics &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;&lt;span class=&quot;token string&quot;&gt;&quot;placeOrder&quot;&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;

6&lt;span class=&quot;token keyword&quot;&gt;public&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;void&lt;/span&gt; &lt;span class=&quot;token function&quot;&gt;listen&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;String orderEntityStr&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

7    log&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;info&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token string&quot;&gt;&quot;------------------ orderEntityStr =&quot;&lt;/span&gt; &lt;span class=&quot;token operator&quot;&gt;+&lt;/span&gt; orderEntityStr&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

8    Gson gs &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;Gson&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

9    OrderEntity orderEntity &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; gs&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;fromJson&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;orderEntityStr&lt;span class=&quot;token punctuation&quot;&gt;,&lt;/span&gt;OrderEntity&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;class&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;&lt;span class=&quot;token comment&quot;&gt;//把JSON字符串转为对象&lt;/span&gt;

10    orderRepository&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;save&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;orderEntity&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

11&lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

12

13

}

接收消息直接用@KafkaListener注解即可，并在监听中设置监听的topic，topics是一个数组所以是可以绑定多个主题的，如@KafkaListener(topics = {“topicA”,“topicB”})。这里的topic需要和消息发送类 KafkaSender.java中设置的topic一致。

spring.kafka.bootstrap-servers 后面设置你安装的Kafka的机器IP地址和端口号9092。

三、启动Kafka服务


1
2
3
1bin/kafka-server-start.sh  config/server.properties

2

3

千万注意：记得将你的虚拟机或者服务器关闭防火墙或者开启Kafka的端口9092。

四、测试

启动order-provider，调用下单接口,可以看到下单成功。
我们来看下Kafka中的topic列表:


1
2
3
1bin/kafka-topics.sh --list --zookeeper localhost:2181

2

3

会看到：


1
2
3
4
1__consumer_offsets

2placeOrder

3

4

接下来，我们来测试下kafka的消费能力。
我们把 OrderService 改造一下：


1
2
3
1package cn.lilyssh.order.provider.service;

2

3

import cn
.lilyssh
.common
.kafka
.provider
.KafkaSender
;

import cn
.lilyssh
.order
.api
.model
.request
.OrderInsertReq
;

import cn
.lilyssh
.order
.api
.service
.OrderServiceApi
;

import cn
.lilyssh
.order
.provider
.dao
.entity
.OrderEntity
;

import com
.alibaba
.dubbo
.config
.annotation
.Service
;

import lombok
.AllArgsConstructor
;

import lombok
.extern
.slf4j
.Slf4j
;

import org
.springframework
.beans
.BeanUtils
;

import org
.springframework
.stereotype
.Component
;

import org
.springframework
.util
.StringUtils
;

import java
.math
.BigDecimal
;

import java
.util
.*
;

@Slf4j

@Service

@Component

@AllArgsConstructor

public
class
OrderService
implements
OrderServiceApi
{


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
1&lt;span class=&quot;token keyword&quot;&gt;private&lt;/span&gt; KafkaSender kafkaSender&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

2

3&lt;span class=&quot;token comment&quot;&gt;/**

4 * 每两秒，新建五百个下单线程，一分钟后停止，查看kafka每秒是否能消费250条数据。

5 */&lt;/span&gt;

6&lt;span class=&quot;token annotation punctuation&quot;&gt;@Override&lt;/span&gt;

7&lt;span class=&quot;token keyword&quot;&gt;public&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;void&lt;/span&gt; &lt;span class=&quot;token function&quot;&gt;saveByKafka&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;OrderInsertReq orderInsertReq&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

8    OrderEntity orderEntity&lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;OrderEntity&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

9    &lt;span class=&quot;token comment&quot;&gt;//直接写入数据库太慢，引起dubbo超时，导致调用多次，此处需要改造成kafka异步写入。&lt;/span&gt;

10    BeanUtils&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;copyProperties&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;orderInsertReq&lt;span class=&quot;token punctuation&quot;&gt;,&lt;/span&gt;orderEntity&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

11    System&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;out&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;println&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token string&quot;&gt;&quot;预备备！开始！&quot;&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

12    Timer timer &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;Timer&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

13    timer&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;schedule&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;MyTask&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;timer&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;,&lt;/span&gt; &lt;span class=&quot;token number&quot;&gt;0&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;,&lt;/span&gt; &lt;span class=&quot;token number&quot;&gt;2000&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;  &lt;span class=&quot;token comment&quot;&gt;//任务等待0秒后开始执行，之后每2秒执行一次&lt;/span&gt;

14&lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

15&lt;span class=&quot;token comment&quot;&gt;//任务：每次新建五百个下单线程。&lt;/span&gt;

16&lt;span class=&quot;token keyword&quot;&gt;class&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;MyTask&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;extends&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;TimerTask&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

17    &lt;span class=&quot;token keyword&quot;&gt;private&lt;/span&gt; Timer timer&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

18    &lt;span class=&quot;token keyword&quot;&gt;public&lt;/span&gt; &lt;span class=&quot;token function&quot;&gt;MyTask&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;Timer timer&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

19        &lt;span class=&quot;token keyword&quot;&gt;this&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;timer &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; timer&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

20    &lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

21

22    &lt;span class=&quot;token keyword&quot;&gt;int&lt;/span&gt; second &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; &lt;span class=&quot;token number&quot;&gt;0&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

23    &lt;span class=&quot;token keyword&quot;&gt;public&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;void&lt;/span&gt; &lt;span class=&quot;token function&quot;&gt;run&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

24        System&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;out&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;println&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token string&quot;&gt;&quot;~~~第&quot;&lt;/span&gt;&lt;span class=&quot;token operator&quot;&gt;+&lt;/span&gt;second&lt;span class=&quot;token operator&quot;&gt;+&lt;/span&gt;&lt;span class=&quot;token string&quot;&gt;&quot;秒~~~&quot;&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

25        &lt;span class=&quot;token keyword&quot;&gt;for&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;int&lt;/span&gt; i &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; &lt;span class=&quot;token number&quot;&gt;0&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt; i &lt;span class=&quot;token operator&quot;&gt;&lt;&lt;/span&gt; &lt;span class=&quot;token number&quot;&gt;500&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt; i&lt;span class=&quot;token operator&quot;&gt;++&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

26            AddOrder addOrder&lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;AddOrder&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

27            Thread thread&lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;Thread&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;addOrder&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

28            thread&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;start&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

29        &lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

30        second&lt;span class=&quot;token operator&quot;&gt;++&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

31        &lt;span class=&quot;token keyword&quot;&gt;if&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt; second &lt;span class=&quot;token operator&quot;&gt;==&lt;/span&gt; &lt;span class=&quot;token number&quot;&gt;30&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

32            &lt;span class=&quot;token keyword&quot;&gt;this&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;timer&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;cancel&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

33            System&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;out&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;println&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token string&quot;&gt;&quot;#### 程序结束 ####&quot;&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

34        &lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

35    &lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

36&lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

37&lt;span class=&quot;token comment&quot;&gt;//下单线程&lt;/span&gt;

38&lt;span class=&quot;token keyword&quot;&gt;class&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;AddOrder&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;implements&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;Runnable&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

39    &lt;span class=&quot;token annotation punctuation&quot;&gt;@Override&lt;/span&gt;

40    &lt;span class=&quot;token keyword&quot;&gt;public&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;void&lt;/span&gt; &lt;span class=&quot;token function&quot;&gt;run&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt; &lt;span class=&quot;token punctuation&quot;&gt;{&lt;/span&gt;

41        OrderEntity orderEntity &lt;span class=&quot;token operator&quot;&gt;=&lt;/span&gt; &lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;OrderEntity&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

42        orderEntity&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;setUserId&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token number&quot;&gt;753&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

43        orderEntity&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;setPayment&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;BigDecimal&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token number&quot;&gt;928.23&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

44        orderEntity&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;setCreateTime&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token keyword&quot;&gt;new&lt;/span&gt; &lt;span class=&quot;token class-name&quot;&gt;Date&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

45        kafkaSender&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;send&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token string&quot;&gt;&quot;placeOrder&quot;&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;,&lt;/span&gt; orderEntity&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;getUserId&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;.&lt;/span&gt;&lt;span class=&quot;token function&quot;&gt;toString&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;(&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;,&lt;/span&gt; orderEntity&lt;span class=&quot;token punctuation&quot;&gt;)&lt;/span&gt;&lt;span class=&quot;token punctuation&quot;&gt;;&lt;/span&gt;

46    &lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

47&lt;span class=&quot;token punctuation&quot;&gt;}&lt;/span&gt;

48

49

}

{{userData.name}}已认证

高并发架构实战(九) Spring Boot集Kafka

一、简介

二、使用方法

（1）添加依赖

（2）在application.yml中添加配置

（3）消息发送类

private Gson gson

（4）在下单业务中调用消息发送

public
void
saveByKafka
(OrderInsertReq orderInsertReq
)
{
OrderEntity orderEntity

（5）消息接收类

三、启动Kafka服务

四、测试

英文站如何做Google Adsense

安全咨询服务

{{userData.name}}已认证

一、简介

二、使用方法

（1）添加依赖

（2）在application.yml中添加配置

（3）消息发送类

private Gson gson

（4）在下单业务中调用消息发送

public void saveByKafka (OrderInsertReq orderInsertReq ) { OrderEntity orderEntity

（5）消息接收类

三、启动Kafka服务

四、测试

Related posts:

英文站如何做Google Adsense

安全咨询服务

.Net Core 分布式微服务框架介绍 - Jimu

架构设计：负载均衡层设计方案（2）——Nginx安装

架构设计：负载均衡层设计方案（8）——负载均衡层总结上篇

LVS负载均衡

public
void
saveByKafka
(OrderInsertReq orderInsertReq
)
{
OrderEntity orderEntity