微服务与消息驱动(RabbitMQ,Kafka)之开发消息微服务

曾经终败给现在 2022-03-08 12:37 288阅读 0赞

上篇概述RabbitMQ和Kafka的入门, 此篇主要概述**Spring could Stream**集成** **RabbitMQ和Kafka 开发消息微服务!

承接上篇 [微服务与消息驱动(RabbitMQ,Kafka)之入门篇][RabbitMQ_Kafka]

**概述:Spring could Stream帮我们做了一定程度的简化,只需少量代码配置就可以实现两个框架的功能,不需要调用API**

**1.准备工作**

首先创建几个项目工程(自行创建),本案例默认使用 [RabbitMQ][RabbitMQ_Kafka]

*  spring-server: Eureka服务器,端口 8761
 *  spring-consumer: 消息消费者, Eureka客户端 ,端口 8080
 *  spring-producer:  消息生产者,Eureka客户端 ,端口 8081

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70][]

**2. 消息生产者:**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 1][]

<dependency>
    			<groupId>org.springframework.cloud</groupId>
    			<artifactId>spring-cloud-starter-config</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.cloud</groupId>
    			<artifactId>spring-cloud-starter-eureka</artifactId>
    		</dependency>
    		
    		<dependency>
    			<groupId>org.springframework.cloud</groupId>
    			<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    		</dependency>

主要引入 **spring-cloud-starter-stream-rabbit** 依赖,引入后会自动帮我们项目引入spring-cloud-stream 以及spring-cloud-stream-binder.如果想使用kafka请将spring-cloud-starter-stream-kafka引入

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 4][]

**3. 消息的消费者:**

**    消费者项目工程所使用的依赖与生产者一致,先了解****生产者与消费者在微服务使用,后面再一起整合测试**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 6][]

依次启动 spring-server(8761端口),spring-producer(8081),spring-consumer(8080),

访问 http://localhost:8081/send

打开消费者控制台,可以看到 hello world 输出,表示消费者已经从消息代理中获取到消息

**大多数情况下我们可以不必编写服务接口,甚至不必使用@input或@output注解,如下图第二个方法**![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 7][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 8][]

为了简化开发 **Spring could Stream 内置了3个接口: Sink,Source和Processor, Processor接口继承于Sink与Source,在实际应用中可以考虑只使用Processor接口,最后给大家演示**

**Sink接口定义源码:**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 9][]

**Source接口定义源码:**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 10][]

**Processor接口定义源码:**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 11][]

**4. 消费者组:**

**    前面讲述kafka框架的时候提到了消费者组的概念,在Spring could Stream中同样引入了这个概念.**

**当消费者组相同时,对于发送过来的消息仅由其中一个消费者实例处理,如果不同则会发送给全部的消费者实例**

**新建两个消费者项目,配置他们的消费者组为groupB,原来的为消费者为****groupA**

*  spring-second-consumer : 端口为8082,消息监听器中会输出 second consumer字符
 *  spring-third-consumer : 端口为8083,消息监听器中会输出 third consumer字符

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 12][]

依次启动server,消息生产者,消息消费者(**groupA和groupB**),启动后访问生产者的发送地址 http://localhost:8081/send

在多吃发送后可以看到,每一次发送的消息原来的消费者都会接收到,后面加的两个消费者则会轮询处理

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 13][]

![20190315162735552.png][]

![20190315162829395.png][]

![20190315162853942.png][]

![20190315162927738.png][]

上面我们说过为了简化代码只需要使用**Processor接口,**下面我们来看一下具体代码

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 14][]

**这里编写的是消费者,与生产者一致需在启动类中绑定消息通道,不需要编写消息通道接口只需要在启动类中加上 Processor.class,在接收或者发送消息时在注解@StreamListener中指定输入消息(Processor.INPUT)还是发送消息(Processor.OUTPUT). 生产者同理可进行重构在此不在重复啰嗦了**

**后续再更新其它内容,敬请关注**

[RabbitMQ_Kafka]: https://blog.csdn.net/lchq1995/article/details/88553880
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70]: /images/20220308/0da63f96376d44d48bd15eed25b17cda.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 1]: /images/20220308/12b49a7c380c47199f79abea84fb79f1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 2]: /images/20220308/8df9a8067d424baca132bb67fd17dd98.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 3]: /images/20220308/1642c2f8a69349cf9037778fc02f5289.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 4]: /images/20220308/ab4d6e3d4a4f4cc8a2ef4a4de82d9d37.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 5]: /images/20220308/846d1514edf2437da71620a319161de8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 6]: /images/20220308/486d5cf070e24d1f87c945cd17c32bba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 7]: /images/20220308/c1c184aa4aa84118a3f707681a9cf923.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 8]: /images/20220308/639a0a0b0d4c48bd9217c773e36058da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 9]: /images/20220308/2cd90472134b4c27ab52c05a654e41c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 10]: /images/20220308/9f06788252cf4164987f2260d875a2fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 11]: /images/20220308/492c4925cd7c44b1baf0d1b6164fb2d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 12]: /images/20220308/002f3cd2e20b47a4b941d8909f043516.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 13]: /images/20220308/d87346d6957d42eea3335c4f59dd2763.png
[20190315162735552.png]: /images/20220308/67e184cf99f34b8688b7a613f14d75f1.png
[20190315162829395.png]: /images/20220308/208b5af7f5c1460f9d84f599a28776ec.png
[20190315162853942.png]: /images/20220308/2ab2b0a8d6e940689b2432119f7199c9.png
[20190315162927738.png]: /images/20220308/ee74fe8127224de28f66324baaa260f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 14]: /images/20220308/b63f755170aa4d14b5f9fc9e853b2030.png