MQ
Message Queue,消息队列,FIFO 结构。
例如电商平台,在用户支付订单后执行对应的操作;
优点:
- 异步
- 削峰
- 解耦
缺点
- 增加系统复杂性
- 数据一致性
- 可用性
JMS
Java Message Service,Java消息服务,类似 JDBC 提供了访问数据库的标准,JMS 也制定了一套系统间消息通信的规范;
ConnectionFactory
Connection
Destination
Session
MessageConsumer
MessageProducer
Message
协作方式图示为;
业界产品
ActiveMQ
作为 Apache 下的开源项目,完全支持 JMS 规范。并且 Spring Boot 内置了 ActiveMQ 的自动化配置,作为入门再适合不过。
快速开始
添加依赖;
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
消息发送;
// 1. 创建连接工厂
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
// 2. 工厂创建连接
Connection connection = factory.createConnection();
// 3. 启动连接
connection.start();
// 4. 创建连接会话session,第一个参数为是否在事务中处理,第二个参数为应答模式
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 5. 根据session创建消息队列目的地
Destination queue = session.createQueue("test-queue");
// 6. 根据session和目的地queue创建生产者
MessageProducer producer = session.createProducer(queue);
// 7. 根据session创建消息实体
Message message = session.createTextMessage("hello world!");
// 8. 通过生产者producer发送消息实体
producer.send(message);
// 9. 关闭连接
connection.close();
Spring Boot 集成
添加依赖;
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
添加 yaml 配置;
spring:
activemq:
broker-url: tcp://localhost:61616
jms:
#消息模式 true:广播(Topic),false:队列(Queue),默认时false
pub-sub-domain: true
收发消息;
@Autowired
private JmsTemplate jmsTemplate;
// 接收消息
@JmsListener(destination = "test")
public void receiveMsg(String msg) {
System.out.println(msg);
}
// 发送消息
public void sendMsg(String destination, String msg) {
jmsTemplate.convertAndSend(destination, msg);
}
高可用
基于 zookeeper 实现主从架构,修改 activemq.xml 节点 persistenceAdapter 配置;
<persistenceAdapter>
<replicatedLevelDB
directory="${activemq.data}/levelDB"
replicas="3"
bind="tcp://0.0.0.0:0"
zkAddress="172.17.0.4:2181,172.17.0.4:2182,172.17.0.4:2183"
zkPath="/activemq/leveldb-stores"
hostname="localhost"
/>
</persistenceAdapter>
broker 地址为:failover:(tcp://192.168.4.19:61616,tcp://192.168.4.19:61617,tcp://192.168.4.19:61618)?randomize=false
负载均衡
在高可用集群节点 activemq.xml 添加节点 networkConnectors;
<networkConnectors>
<networkConnector uri="static:(tcp://192.168.0.103:61616,tcp://192.168.0.103:61617,tcp://192.168.0.103:61618)" duplex="false"/>
</networkConnectors>
集群消费
由于发布订阅模式,所有订阅者都会接收到消息,在生产环境,消费者集群会产生消息重复消费问题。
ActiveMQ 提供 VirtualTopic 功能,解决多消费端接收同一条消息的问题。于生产者而言,VirtualTopic 就是一个 topic,对消费而言则是 queue。
在 activemq.xml 添加节点 destinationInterceptors;
<destinationInterceptors>
<virtualDestinationInterceptor>
<virtualDestinations>
<virtualTopic name="testTopic" prefix="consumer.*." selectorAware="false"/>
</virtualDestinations>
</virtualDestinationInterceptor>
</destinationInterceptors>
生产者正常往 testTopic 中发送消息,订阅者可修改订阅主题为类似 consumer.A.testTopic 这样来消费。
RocketMQ
是一个队列模型的消息中间件,具有高性能、高可靠、高实时、分布式特点。
架构图示
Name Server
名称服务器,类似于 Zookeeper 注册中心,提供 Broker 发现;
Broker
RocketMQ 的核心组件,绝大部分工作都在 Broker 中完成,接收请求,处理消费,消息持久化等;
Producer
消息生产方;
Consumer
消息消费方;
快速开始
添加依赖;
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.5.2</version>
</dependency>
消息发送;
DefaultMQProducer producer = new DefaultMQProducer("producer-group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.setInstanceName("producer");
producer.start();
Message msg = new Message(
"producer-topic",
"msg",
"hello world".getBytes()
);
//msg.setDelayTimeLevel(1);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult.toString());
producer.shutdown();
消息接收;
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.setInstanceName("consumer");
consumer.subscribe("producer-topic", "msg");
consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
Spring Boot 集成
添加依赖;
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.4</version>
</dependency>
添加 yaml 配置;
rocketmq:
name-server: 127.0.0.1:9876
producer:
group: producer
发送消息;
@Autowired
private RocketMQTemplate mqTemplate;
public void sendMessage(String topic, String tag, String message) {
SendResult result = mqTemplate.syncSend(topic + ":" + tag, message);
System.out.println(JSON.toJSONString(result));
}
接收消息;
@Component
@RocketMQMessageListener(consumerGroup = "consumer", topic = "topic-test", selectorExpression = "tag-test")
public class MsgListener implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
System.out.println(message);
}
}
Console 控制台
RocketMQ 拓展包提供了管理控制台;
https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console
重复消费
产生原因:
- 生产者重复投递;
- 消息队列异常;
- 消费者异常消费;
怎么解决重复消费的问题,换句话怎么保证消息消费的幂等性。
顺序消息
消息错乱的原因:
- 一个消息队列 queue,多个 consumer 消费;
- 一个 queue 对应一个 consumer,但是 consumer 多线程消费;
要保证消息的顺序消费,有三个关键点:
- 消息顺序发送
- 消息顺序存储
- 消息顺序消费
分布式事务
在分布式系统中,一个事务由多个本地事务组成。这里介绍一个基于 MQ 的分布式事务解决方案。
通过 broker 的 HA 高可用,和定时回查 prepare 消息的状态,来保证最终一致性。