SpringCloud stream消息驱动和链路追踪

简要说明

在学习消息驱动之前，肯定有之前并没有接触过rabbitMQ或者其他的MQ（消息中间件）的。为了保证springcloud系列学习的连续性为之后的springcloud alibaba做好准备，只能先提前简单的学习一下rabbitMQ的安装。保证整个学习的完整性。之后打算在学习一下中间件。重新温习一下redis等之后。作为普通学生今后连续性学习编程的时间可能就不多了。毕竟还有考研深造的打算。在在时间上总是要放弃一些的。个人也会坚持每周抽时间学一下。既是为了防止遗忘的太多。同时也是为了自己的兴趣。如有学习建议欢迎指导

RabbitMQ安装

RabbitMQ和redis一样安装在Linux的虚拟机上效果会好一些。因为没有系统的学习过，这一次我选择安装window版本的RabbitMQ

在系统学习的时候在去使用centos来进行安装。

RabbitMQ是用Erlang来编写的需要Erlang的环境，现在自己的电脑上安装Erlang的环境。

安装环境的下载地址：https://www.erlang.org/downloads

选择最新版本的windows的版本进行安装。点击Next

选择好安装位置以后进行安装。

和java一样配置erlang语言的环境变量ERLANG_HOME

之后在去配置path变量 将%ERLANG_HOME%\bin加入到path中。完成之后进行保存。保存之后验证环境变量是否配置成功。

在cmd中使用erl测试环境变量是否配置成功。

依赖的环境完成之后在安装window版的rabbitMQ

下载并安装RabbitMQ

• 下载地址：http://www.rabbitmq.com/download.html

进入官网的指定位置来下载RabbitMQ

在官方的网址上如果找不到可以用下面的地址从github上下载旧的版本

https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.7/rabbitmq-server-3.7.7.exe

点击Next选择对应的安装位置。finsh之后下一步就是来配置RabbitMQ了

配置RabbitMQ

配置RabbitMQ的环境变量

启动 rabbitmq_management

在命令行CMD中使用dos命令，进入rabbitmq安装目录的sbin

执行以下命令：进行安装

rabbitmq-plugins enable rabbitmq_management 

打开命令行命令行,进入RabbitMQ的安装目录: sbin

输入 rabbitmqctl status , 如果出现以下的图，说明安装是成功的，并且说明现在RabbitMQ Server已经启动了,运行正常。

在安装完成之后我们打开sbin目录，双击rabbitmq-server.bat启动服务看看是否可以进行访问。

启动RabbitMQ

之后在浏览器输入http://localhost:15672/即可访问成功。

用户名 密码均为: guest, 登录

至此, windows安装RabbitMQ已全部完成

在安装完window版本的rabbitMQ之后便可以去完成springcloud的消息驱动的学习了，之后的开发中还是建议将RabbitMQ安装到Linux虚拟机上。

Stream消息驱动简单介绍

为什么要引入SpringCloud Stream？

技术兴起的原因：为了解决系统中不同中间件的适配问题，出现了cloud stream，采用适配绑定的方式，自动给不同的MQ之间进行切换。

屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型。

举例：对于我们Java程序员来说，可能有时要使用ActiveMQ,有时要使用RabbitMQ,甚至还有RocketMQ以及Kafka，这之间的切换似乎很麻烦，我们很难，也没有太多时间去精通每一门技术，那有没有一种新技术的诞生，让我们不再关注具体MQ的细节，自动的给我们在各种MQ内切换。

简介：Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。它可以基于 Spring Boot 来创建独立的、可用于生产的 Spring 应用程序。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，并引入了发布-订阅、消费组、分区这三个核心概念。通过使用 Spring Cloud Stream，可以有效简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。但是目前 Spring Cloud Stream 只支持 RabbitMQ 和 Kafka 的自动化配置。

一句话：屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型。

Binder Implementations 绑定器
通过绑定器Binder作为中间件，实现了应用程序与消息中间件细节的解耦。

Input对应消息生产者

Output对应消息消费者

相关术语

MQ相关术语
Message：生产者/消费者之间靠消息媒介传递信息内容

MessageChannel：消息必须走特定的通道

消息通道的子接口SubscribableChannel，由MessageHandle消息处理器所订阅。

相关注解
Middleware：中间件，目前只支持RabbitMQ和Kafka
Binder：应用层和消息中间件之间的封装，实现了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变。消息类型，这些可以通过配置文件修改。
Input：表示输入通道，消息进入该通道传到应用程序。
Output：注解标识输出通道，发布的消息将通过该通道离开应用程序。
StreamListener：监听队列，用于消费者的队列的消息接收。
EnableBinding：将信道channel和exchange绑定在一起。

Stream消息驱动生产者

新建子模块

cloud-stream-rabbitmq-provider8801 消息生产模块

修改yaml配置文件

binder: defaultRabbit 这里报红,但是不影响操作

server:port: 8801spring:application:name: cloud-stream-providercloud:stream:binders: # 在此处配置要绑定的rabbitmq的服务信息；defaultRabbit: # 表示定义的名称，用于于binding整合type: rabbit # 消息组件类型environment: # 设置rabbitmq的相关的环境配置spring:rabbitmq:host: localhost #rabbitmq服务启动所在机器的IP地址port: 5672username: guestpassword: guestbindings: # 服务的整合处理output: # 这个名字是一个通道的名称destination: studyExchange # 表示要使用的Exchange名称定义content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka:client: # 客户端进行Eureka注册的配置service-url:defaultZone: http://localhost:7001/eurekainstance:lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）instance-id: send-8801.com  # 在信息列表时显示主机名称prefer-ip-address: true     # 访问的路径变为IP地址

创建启动类

package com.dzu.springcloud;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class StreamMQMain8801 {public static void main(String[] args) {SpringApplication.run(StreamMQMain8801.class,args);}
}

创建业务类

包括三个部分发送消息的接口，发送消息接口的实现类，controller

package com.dzu.springcloud.service;public interface IMessageProvider {String sendMessage();
}

package com.dzu.springcloud.service.impl;import com.dzu.springcloud.service.IMessageProvider;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;import javax.annotation.Resource;
import java.util.UUID;@EnableBinding(Source.class) //定义消息的推送管道
public class IMessageProviderImpl implements IMessageProvider
{@Resourceprivate MessageChannel output; // 消息发送管道@Overridepublic String sendMessage(){String serial = UUID.randomUUID().toString();output.send(MessageBuilder.withPayload(serial).build());System.out.println("*****serial: "+serial);return null;}
}

package com.dzu.springcloud.controller;import com.dzu.springcloud.service.IMessageProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class SendMessageController {@Autowiredprivate IMessageProvider iMessageProvider;@GetMapping(value = "/sendMessage")public String send(){return  iMessageProvider.sendMessage();}
}

之后打开window版本的rabbitMQ进行测试。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GZTguW2z-1642512161927)(C:\Users\Lenovo\AppData\Roaming\Typora\typora-user-images\image-20220117205905187.png)]

Stream消息驱动消费者

新建cloud-stream-rabbitmq-consumer8802消息驱动的消费者模块收到对应消息

引入和消费者相同的pom依赖

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

修改pom文件和上面生产者的配置信息相同。（只修改端口号和input）

server:port: 8802spring:application:name: cloud-stream-providercloud:stream:binders: # 在此处配置要绑定的rabbitmq的服务信息；defaultRabbit: # 表示定义的名称，用于于binding整合type: rabbit # 消息组件类型environment: # 设置rabbitmq的相关的环境配置spring:rabbitmq:host: localhost #rabbitmq服务启动所在机器的IP地址port: 5672username: guestpassword: guestbindings: # 服务的整合处理input: # 这个名字是一个通道的名称destination: studyExchange # 表示要使用的Exchange名称定义content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka:client: # 客户端进行Eureka注册的配置service-url:defaultZone: http://localhost:7001/eurekainstance:lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）instance-id: send-8802.com  # 在信息列表时显示主机名称prefer-ip-address: true     # 访问的路径变为IP地址

创建该模块的主启动类

package com.dzu.springcloud;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class StreamMQMain8802 {public static void main(String[] args) {SpringApplication.run(StreamMQMain8802.class,args);}
}

编写核心的业务逻辑的代码

package com.dzu.springcloud.controller;import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {@Value("${server.port}")private String serverPort;@StreamListener(Sink.INPUT)public void input(Message<String> message){System.out.println("消费者1号，------>接收到的消息： "+message.getPayload()+"\t"+" port: "+serverPort);}
}

启动8802测试是否可以接收到传来的消息。

分组消费和持久化

1、仿照8802，克隆一份8803

2、启动项目

RabbitMQ、7001（服务注册）、8801（消息生产）、8802（消息消费）、8803（消息消费）

3、运行后的两个问题

1、有重复消费问题
2、消息持久化问题

先开看重复消费的问题。（两个消费者微服务同时收到相同的重复信息）

目前是8802/8803同时都收到了，存在重复消费问题
比如在如下场景，订单系统我们做集群部署，都会从RabbitMQ中获取订单信息，

那如果一个订单同时被两个服务获取到，那么就会造成数据错误，我们得避免这种情况。

这时我们就可以使用Stream中的消息分组来解决

注意在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。
不同组件是可以全面消费的（重复消费）
同一个组内发生竞争关系，只有其中一个可以消费

解决消息重复消费问题

首先自定义分组（修改配置文件）

在8802和8803的配置文件的下面都加上相同的分组

group: spectrumrpcA

server:port: 8802spring:application:name: cloud-stream-providercloud:stream:binders: # 在此处配置要绑定的rabbitmq的服务信息；defaultRabbit: # 表示定义的名称，用于于binding整合type: rabbit # 消息组件类型environment: # 设置rabbitmq的相关的环境配置spring:rabbitmq:host: localhost #rabbitmq服务启动所在机器的IP地址port: 5672username: guestpassword: guestbindings: # 服务的整合处理input: # 这个名字是一个通道的名称destination: studyExchange # 表示要使用的Exchange名称定义content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”binder: defaultRabbit # 设置要绑定的消息服务的具体设置group: spectrumrpcAeureka:client: # 客户端进行Eureka注册的配置service-url:defaultZone: http://localhost:7001/eurekainstance:lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）instance-id: send-8802.com  # 在信息列表时显示主机名称prefer-ip-address: true     # 访问的路径变为IP地址

在配置了相同分组之后同一个组中的不同微服务之间采用轮询的算法类接收消息。

配置了分组之后就可以解决重复消费和持久化的问题。

Sleuth链路追踪

定义：springcloud Slenuth中提供了一套完整的服务跟踪解决方案。在分布式系统中提供了解决方案并兼容支持了zipkin

搭建链路监控

官网下载zipkin的jar包

https://zipkin.io/pages/quickstart.html

cmd命令行java -jar运行该jar包之后在通过浏览器访问

引入了zipkin的jar包之后通过仪表盘显示即可。

具体的配置可以参考其他的参考资料不在进行详细说明。

总结：

到此springcloud微服务框架的一站式开发已经学习完成，之后便可以进行springcloud alibaba的测试和学习了。