消息队列Mq是个好东西

一、前言

在web 系统中，系统与系统的交互中介除了http的三次握手，有个重要的组件经常用到那就是消息队列。消息队列以牺牲少许得到实时性，却解决了很大的问题。

消息队列

二、消息队列的使用场景

系统解耦

开闭原则是设计模式中的一项原则，意味着软件实体应该对扩展开放、对修改关闭，以保持系统的独立性。这原则强调解耦思想，通过消息队列的使用，我们可以看作在系统中隐含地引入了一个对外的扩展接口。这样，可以方便地解耦业务逻辑，调用者只需发送消息，而不必关心下游逻辑如何执行。消息队列的存在使得系统的各个组件能够通过发布-订阅模式进行通信，从而增加了系统的可扩展性。

很多同学可能有疑问，系统之间的解耦，使用 RPC 服务调用也可以实现，使用消息队列有什么好处吗？使用远程服务调用，需要在其中一个调用方进行显式地编码业务逻辑；如果使用消息队列就不会有这个问题了，系统之间可以更好地实现依赖倒转，这也是设计模式中的一个重要原则。

异步处理

异步处理是在处理高并发、高可用系统设计中的重要机制之一。当系统无法立即处理消息，或受到系统承载能力限制时，可以应用消息队列将请求进行异步化处理。

一个典型的异步处理场景是流量削峰。以电商的秒杀场景为例，秒杀活动常常引发高峰期的访问流量。然而，服务往往无法瞬时承受如此高的访问压力。为了解决这个问题，可以引入消息队列，并结合流量控制工具，将超过系统阈值的请求存放在消息队列中，待流量高峰过去后再进行处理。

通过异步处理的方式，系统能够更好地应对高并发的请求，避免因流量过大而导致的系统崩溃或性能下降。消息队列起到了缓冲和调节流量的作用，有效平衡了系统的负载。

请求缓冲

在生产者和消费者模型中，可以使用消息队列作为缓冲层。消息队列可以平滑处理不同业务系统之间的处理性能差异。在早期的企业应用系统中，存在一个企业数据总线（ESB）的概念，用于实现内部各个系统的集成。

数据分发

消息队列提供了不同的订阅模式，其中一对多的广播机制是其中之一，它可以用来实现数据的分发。一个典型的例子是关系型数据库对于binlog的订阅处理。

在关系型数据库中，binlog是指记录数据库变更的日志。主库生成的binlog只有一份，但下游的消费方可能包括各种文件索引、离线数据库等。这时候，可以应用消息队列来实现数据的分发。

除了这些典型应用，消息队列还可以用来实现分布式事务，前面“分布式事务的解决方案”中我们提过，利用数据库+本地消息表的方式分布式一致性，是一个非常经典的分布式事务解决方案。

三、介绍rocketMQ

这里已腾讯云rocketMQ为例

在maven pom文件引入依赖

		<!--cos -->
		<dependency>
			<groupId>com.qcloud</groupId>
			<artifactId>cos_api</artifactId>
			<version>${cos_api.version}</version>
		</dependency>
		
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>${rocketmq.version}</version>
        </dependency>
        
		<dependency>
			 <groupId>org.apache.rocketmq</groupId>
			 <artifactId>rocketmq-acl</artifactId>
			 <version>${rocketmq.version}</version>
		</dependency>
		
 

import java.io.UnsupportedEncodingException;
import java.util.List;

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import lombok.extern.slf4j.Slf4j;
 
@Slf4j
@Service
@Transactional(rollbackFor = Exception.class)
public class RocketTXMqService {

	@Value("${rocketmq.namespace:-1}")
	private String namespace;
	@Value("${rocketmq.producer.group:-1}")
	private String groupName;
	@Value("${rocketmq.producer.access-key:-1}")
	private String accessKey;
	@Value("${rocketmq.producer.secret-key:-1}")
	private String secretKey;
	@Value("${rocketmq.name-server:-1}")
	private String nameserver;

	// MQ生产者
	private DefaultMQProducer producer;

	// MQ实例化消费者push
	private DefaultMQPushConsumer pushConsumer;

	// MQ实例化消费者pull
	private DefaultLitePullConsumer pullConsumer;

	/**
	 * 创建生产者
	 * 
	 * @return
	 */
	public DefaultMQProducer getProducer() {

		if (null == producer) {
			// 实例化消息生产者Producer
			producer = new DefaultMQProducer(namespace, groupName,
					new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) // ACL权限
			);
			// 设置NameServer的地址
			producer.setNamesrvAddr(nameserver);
			try {
				// 启动Producer实例
				producer.start();
			} catch (MQClientException e) {
				e.printStackTrace();
			}
		}
		return producer;
	}

	/**
	 * 同步发送 发送消息
	 */
	public void syncSend(String topic, String tag, String data) {
		producer = getProducer();
		// 发送消息
		SendResult sendResult = null;
		try {
			// 创建消息实例，设置topic和消息内容
			Message msg = new Message(topic, tag, data.getBytes(RemotingHelper.DEFAULT_CHARSET));
			sendResult = producer.send(msg);
			log.info("埋点信息发送腾讯云MQ：" + data);
			log.info("发送腾讯云MQ接口返回状态sendResult:" + sendResult);
		} catch (UnsupportedEncodingException e) {
			log.error("UnsupportedEncodingException:" + e.getMessage());
		} catch (MQClientException e) {
			log.error("MQClientException:" + e.getMessage());
		} catch (RemotingException e) {
			log.error("RemotingException:" + e.getMessage());
		} catch (MQBrokerException e) {
			log.error("MQBrokerException:" + e.getMessage());
		} catch (InterruptedException e) {
			log.error("InterruptedException:" + e.getMessage());
		}
	}

	/**
	 * 创建push消费者
	 * 
	 * @return
	 */
	public DefaultMQPushConsumer getPushConsumer() {

		if (null == pushConsumer) {// 实例化消费者
			pushConsumer = new DefaultMQPushConsumer(namespace, groupName,
					new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); // ACL权限
			// 设置NameServer的地址
			pushConsumer.setNamesrvAddr(nameserver);
		}
		return pushConsumer;
	}

	/**
	 * 创建pull 消费者
	 * 
	 * @return
	 */
	public DefaultLitePullConsumer getPullConsumer() {

		if (null == pullConsumer) {// 实例化消费者
			// 实例化消费者
			pullConsumer = new DefaultLitePullConsumer(namespace, groupName,
					new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
			// 设置NameServer的地址
			pullConsumer.setNamesrvAddr(nameserver);
			// 设置从第一个偏移量开始消费
			pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		}
		return pullConsumer;
	}

	/**
	 * push方式订阅消费
	 * 
	 * @param topicName
	 */
	public void pushConsumer(String topicName) {

		pushConsumer = this.getPushConsumer();
		if (null != pushConsumer) {

			try {
				pushConsumer.subscribe(topicName, "*");
				// 注册回调实现类来处理从broker拉取回来的消息
				pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
					// 消息处理逻辑
					log.info("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
					// 标记该消息已经被成功消费, 根据消费情况，返回处理状态
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				});
				// 启动消费者实例
				pushConsumer.start();
			} catch (MQClientException e) {
				log.error("push MQClientException:" + e.getMessage());
			}
		}
	}

	/**
	 * pull方式订阅消费
	 * 
	 * @param topicName
	 */
	public void pullConsumer(String topicName) {
		pullConsumer = this.getPullConsumer();

		if (null != pullConsumer) {
			try {
				// 订阅topic
				pullConsumer.subscribe(topicName, "*");
				// 启动消费者实例
				pullConsumer.start();
			} catch (MQClientException e) {
				log.error(" pull MQClientException:" + e.getMessage());
			}
			try {
				log.info("Consumer Started.%n");
				while (true) {
					// 拉取消息
					List<MessageExt> messageExts = pullConsumer.poll();
					log.info("%s%n", messageExts);
				}
			} finally {
				pullConsumer.shutdown();
			}
		}
	}

}

向mq中发送消息

?@Autowired
?private RocketTXMqService rocketTXMqService;

String input = JSONObject.toJSONString(dataParam);

rocketTXMqService.syncSend("buried-point-topic", mAccessLog.getEventAliasName(), input);