失眠网 > springcloud微服务架构开发实战：分布式消息总线

springcloud微服务架构开发实战：分布式消息总线

时间：2021-11-22 06:46:49

消息总线的定义

前面在1.4.2节中强调过，在微服务架构中，经常会使用REST 服务或基于消息的通信机制。

在3.6节中也详细介绍了消息通信的实现方式。消息总线就是一种基于消息的通信机制。

消息总线是一种通信工具，可以在机器之间互相传输消息、文件等，它扮演着—种消息路由的角色，拥有一套完备的路由机制来决定消息传输方向。发送端只需要向消息总线发出消息，而不用管消息被如何转发。

Spring Cloud Bus通过轻量消息代理连接各个分布的节点。管理和传播所有分布式项目中的消息，本质是利用了MQ的广播机制在分布式的系统中传播消息，目前常用的有Kafka和RabbitMQ等。

消息总线常见的设计模式

在消息总线中，常见的设计模式有点对点模式及订阅/发布模式。

1.点对点(P2P)

点对点模式包含三个角色。

消息队列（ Queue )。生产者（ Producer ) 。消费者（Consumer )。

点对点模式中的每个消息都被发送到一个特定的队列，消费者从队列中获取消息。队列保留着消息，直到它们被消费或超时。图16-1展示了点对点模式的运行流程图。

点对点模式具有以下特点。

每个消息只有一个消费者，即消息一旦被消费，就不在消息队列中了。生产者和消费者之间在时间上没有依赖性，也就是说当生产者发送了消息之后，不管消费者有没有正在运行，都不会影响到消息被发送到队列。消费者在成功接收消息之后需向队列应答成功，这样消息队列才能知道消息是否被成功消费。

2.订阅/发布（PublSub )

订阅/发布模式包含三个角色。

主题（Topic )。发布者（Publisher )。订阅者（Subscriber )。

订阅/发布模式中，多个发布者将消息发送到对应的主题，系统将这些消息传递给多个订阅者。图16-2展示了订阅/发布模式的运行流程图。

订阅/发布模式具有以下特点。

每个消息可以有多个消费者。主题可以被认为是消息的传输中介，发布者发布消息到主题，订阅者从主题订阅消息。主题使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。

消息总线的意义

在微服务架构中，经常会使用REST服务作为服务间的通信机制。REST以其轻量、简单、易理解而著称，但这种通信机制也并非适合所有的场景。例如，在一些高并发、高可靠、实时的场景，则需要消息总线来帮忙。

概括起来，消息总线具有以下几个优点。

1.实时性高

与REST 服务的“请求—响应”模式不同，消息总线的实时性非常高。使用了消息总线，生产者一方只要把消息往队列里一扔，就可以立马返回，响应用户了。无须等待处理结果，实现了异步处理。

同时，对于消费者而言，消费者对于消息的到达感知也非常及时。消费者会对消息总线进行监听，只要有消息进入队列，就可以马上得到通知。这种优势是REST 服务所不能具备的。在REST服务中，要想及时获取到更新通知，就不得不进行轮询。这往往非常低效。

2生产者与消费者解耦

在消息总线中，生产者负责将消息发送到队列中，而消费者把消息从队列中取出来。生产者无须等待消费者启动，消费者也无须关心生产者是否已经处于就绪状态。所以，这种模式能很好地实现生产者与消费者的解耦。

然而，如果是在REST服务中，服务调用方必须等待服务的提供方准备好了才能调用，否则就会调用失败。

3.故障率低

消息总线拥有对其他通信方式更高的成功率。一方面，生产者与消费者之间实现了解耦，所以，生产者与消费者之间不存在强关联关系，即便是生产者或消费者任意一方掉线了，也不会影响消息最终的送达;另一方面，消息总线往往会结合数据库来实现消息的持久化，并设置状态标识。只有消息消费成功，才会去修改状态标识。

消息总线同时还承担着缓冲区的作用。大量业务消息首先会进入消息队列进行缓存，消息的消费者可以根据自己的处理能力来进行消费，所以不管消息的数据量有多少，都不会对消费者造成冲击。

消息总线常见的实现方式

《分布式系统常用技术及案例分析》一书列举了非常多的流行的、开源的分布式消息服务，如Apache ActiveMQ、RabbitMQ、Apache RocketMQ、Apache Kafka等。这些消息中间件都实现了点对点模式及订阅/发布模式等常见的消息模式。

以下例子演示的是使用ActiveMQ实现生产者—消费者的Java实现方式。

生产者程序Producer.java:

public class Producer{private static final Logger LOGGER=LoggerFactory.getLogger (Producer.class）;private static final string BROKER_URE = ActiveMQConnection.DEFAULT_BROKER URL;private static final String SUBJECT= "waylau-queue";public static void main (String[] args) throws JMSException f//初始化连接工厂ConnectionFactory connectionFactory= new ActiveMQConnectionFactory(BROKER_URL）;//获得连接Connection conn = connectionFactory.createConnection();//启动连接conn.start(;//创建session，第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE）;//创建队列Destination dest = session.createQueue(SUBJECT);//createTopic方法用来创建Topic//session.createTopic ("TOPIC"）;//通过session 可以创建消息的生产者MessageProducer producer = session.createProducer(dest);for(int i=0;i<100;i++）{//初始化一个MQ消息TextMessage message= session.createTextMessage ("Welcome "+i）;//发送消息producer. send(message）;LOGGER.info("send message {}",i）;//关闭 MQ 连接conn.close(）;}}

消费者程序Consumer.java:

public class Consumer implements MessageListener {private static finalLogger LOGGER = LoggerFactory.getLogger(Consumer.class);private static final String BROKER_URL = ActiveMQConnection.DEFAULTBROKER URL;private static final string SUBJECT = "waylau-queue";public static void main(String[] args) throws JMSExceptionf//初始化 ConnectionFactoryConnectionFactory connectionFactory =new ActiveMOConnectionFactory(BROKER_URL）;//创建Mo连接Connection conn = connectionFactory.createConnection();//启动连接conn .start(;//创建会话Session session= conn.createSession (false,Session.AUTO_ACKNOWLEDGE）;//通过会话创建目标Destination dest = session.createQueue(SUBJECT);//创建 MO 消息的消费者MessageConsumer consumer = session.createConsumer(dest);//初始化 MessageListenerconsumer me=newConsumer(）;//给消费者设定监听对象consumer .setMessageListener (me）;@overridepublic void onMessage(Message message){TextMessage txtMessage =(TextMessage)message;try{LOGGER.info("get message " + txtMessage.getText());}catch (JMSException e) {LOGGER.error("error {}",e）);}}

执行命令来启动ActiveMQa:

bin/activemg start

生产者执行如下命令:

mvn clean compile exec:java -Dexec.mainClass=com.waylau.activemq.ProducerApp

输出如下。

20:12:10.807 [ActiveMQ Task-1]INEO org.apache.activemq.transport.failover.FailoverTransport- Successfully connected to tcp://localhost:6161620:12:10.928[main] INFOcom.waylau.activemq.Producer- send message 020:12:10.963 [main] INPO com.waylau.activemq.Producer- send message 120:12:10.992 [main] INFO com.waylau.activemq.Producer - send message 220:12:11.019[main] INFO com.waylau.activemq.Producer - send message 320:12:11.036[main] INFOcom.waylau.activemq.Producer- send message 420:12:11.058 [main] INFO com.waylau.activemq.producer -send message 520:12:11.085[main] INFOcom.waylau.activemq.Producer - send message620:12:11.113 [main] INFOcom.waylau.activemq.Producer - send message 720:12:11.141[main] INFOcom.waylau.activemq.Producer - send message 820:12:11.191 [main] INFO com.waylau.activemq.Producer- send message 9

消费者执行如下命令:

mvn clean compile exec:java-Dexec.mainClass=com.waylau.activemq. ConsumerApp

输出如下。

20:12:05.262[ActiveMQ Task-1] INFO org.apache.activemq.transport.failover.FailoverTransport- Successfully connected to tcp://localhost:6161620:12:10.875 [ActiveMQ Session Task-1] INEOcom.waylau.activemg.Consumer -get message welcome to o20:12:10.939 [ActiveMQ Session Task-1]INFO com.waylau.activemq.Consumer-get message welcome to 120:12:10.965 [ActiveMQ Session Task-1] INFO com.waylau.activemq.Consumer-get message Welcome to 220:12:10.994 [ActiveMQ Session Task-1] INFO com.waylau.activemq. Consumer -get message Welcome to waylau .com 320:12:11.020 [ActiveMQ Session Task-1] INFO com.waylau.activemq. Consumer-get message Welcome to 420:12:11.038 [ActiveMQ Session Task-1] INFO com.waylau.activemq.Consumer-get message Welcome to 520:12:11.059 [ActiveMQ Session Task-1] INFO com.waylau.activemq. Consumer -get message Welcome to 620:12:11.086[ActiveMQ Session Task-1] INEO com.waylau.activemq. Consumer-get message welcome to 720:12:11.114[ActiveMQ Session Task-1] INFO com.waylau.activemq.Consumer-get message Welcome to 820:12:11.142 [ActiveMQ Session Task-1] INFO com.waylau.activemq. Consumer-get message Welcome to 9

上述例子的源码，可以在

/waylau/distributed-systems-technologies-and-cas-es-analysis网址的samples目录下找到。