消息队列
消息队列
前言：
说实话，最近还是比较忙的，手上素材倒是一大把，但是大多只是初步整理了。但是博客这种东西还是要写的，果然后面还是要放低一下排版要求（扩展性的一些东西也少提一些）。
简介：
消息队列这个东西，其实网上的资料还是很多的。我就简单说一些自己的认识与源代码哈。
演变：
我是很喜欢了解技术演进的，因为演进的过程展现了前辈们的智慧。
最早的程序串行执行就不说了。
程序调用中的方法调用，往往调用方与被调用方都存在与同一内存空间（从Java角度说，都是在同一JVM中），所以方法调用的逻辑不会太复杂。简单来说，就是调用方（Java中其实就是目标对象）将被调用方压入Java虚拟机栈，从而执行（详见JVM）。或者等我什么时候，把我有关JVM的笔记贴出来（嘿嘿）。
后来呢，就是出现了对非本地JVM方法调用的需求（举个例子，我需要调用第三方的方法，如果每次都要双方都写一个专门的处理服务（在当时，也许接口更为准确），比较麻烦），那么就有了RPC与RMI的一个需要。那么在Java中就出现了一个stub的技术，定义好后，相关方法就像调用本地一样（详见《Head First Java》相关章节）。当然了，这个时候已经有了中间件的概念了，所以也就有了CORBA等框架。谈到中间件，感兴趣的，可以去查询一下当时主流的中间件分类（如RPC，RMI，MOM，TPM，ORB）。
那么到了现在呢，分布式系统的通信可以按照同步与异步分为两大支柱。之所以这么理解，是因为分布式系统往往同步通信与异步通信都是需要的。简单提一下，同步通信业务逻辑相对简单，实现快速，可以实时获得回应，但耦合度较高。异步通信耦合度低，并可以进行消息堆积，消峰，但无法实时获取回应，业务逻辑复杂，从而提高系统复杂度（尤其当一条业务线与多层异步逻辑）等。之后有机会，我会举例细述。
当然了，在本篇中，只简单谈一下异步通信的主流实现-消息队列。
选择：
选择方面，我就不多说了，目前只用过RabbitMq,RocketMq,Kafka。网上有关消息队列选择的文章很多，很细致，我就不赘述了。
代码实现：
这里贴出来的都是实际生产代码（如果内部版本也算的话，嘿嘿），所以如果有一些不是很熟悉的类，请查看import，是否是项目自身的类。或者也可以直接询问我。
初步实现：
这里的初步实现，是根据RabbitMq的原生方法进行编写（详细参考：《RabbitMQ实战指南》第一章的两个代码清单及第二章的相关解释）。
producer:
package com.renewable.gateway.rabbitmq.producer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import com.renewable.gateway.pojo.Terminal;
import com.renewable.gateway.util.JsonUtil;
import com.renewable.gateway.util.PropertiesUtil;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import static com.renewable.gateway.common.constant.RabbitmqConstant.*;
/**
* @Description：
* @Author: jarry
*/
@Component("TerminalProducer")
public class TerminalProducer {
private static final String IP_ADDRESS = PropertiesUtil.getProperty(RABBITMQ_HOST);
private static final int PORT = Integer.parseInt(PropertiesUtil.getProperty(RABBITMQ_PORT));
private static final String USER_NAME = PropertiesUtil.getProperty(RABBITMQ_USER_NAME);
private static final String USER_PASSWORD = PropertiesUtil.getProperty(RABBITMQ_USER_PASSWORD);
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE = "exchange-terminal-config-terminal2centcontrol";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE = "queue-terminal-config-terminal2centcontrol";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINETYPE = "topic";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_BINDINGKEY = "terminal.config.terminal2centcontrol";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINGKEY = "terminal.config.terminal2centcontrol";
public static void sendTerminalConfig(Terminal terminal) throws IOException, TimeoutException, InterruptedException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(IP_ADDRESS);
factory.setPort(PORT);
factory.setUsername(USER_NAME);
factory.setPassword(USER_PASSWORD);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINETYPE, true, false, null);
channel.queueDeclare(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE, true, false, false, null);
channel.queueBind(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_BINDINGKEY);
String terminalStr = JsonUtil.obj2String(terminal);
channel.basicPublish(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINGKEY, MessageProperties.PERSISTENT_TEXT_PLAIN, terminalStr.getBytes());
channel.close();
connection.close();
}
}
consumer:
package com.renewable.gateway.rabbitmq.consumer;
import com.rabbitmq.client.*;
import com.renewable.gateway.common.GuavaCache;
import com.renewable.gateway.common.ServerResponse;
import com.renewable.gateway.pojo.Terminal;
import com.renewable.gateway.service.ITerminalService;
import com.renewable.gateway.util.JsonUtil;
import com.renewable.gateway.util.PropertiesUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import static com.renewable.gateway.common.constant.CacheConstant.TERMINAL_MAC;
import static com.renewable.gateway.common.constant.RabbitmqConstant.*;
/**
* @Description：
* @Author: jarry
*/
@Component
public class TerminalConsumer {
@Autowired
private ITerminalService iTerminalService;
private static final String TERMINAL_CONFIG_CENTCONTROL2TERMINAL_EXCHANGE = "exchange-terminal-config-centcontrol2terminal";
private static final String TERMINAL_CONFIG_CENTCONTROL2TERMINAL_QUEUE = "queue-terminal-config-centcontrol2terminal";
private static final String TERMINAL_CONFIG_CENTCONTROL2TERMINAL_ROUTINETYPE = "topic";
private static final String TERMINAL_CONFIG_CENTCONTROL2TERMINAL_BINDINGKEY = "terminal.config.centcontrol2terminal";
@PostConstruct
public void messageOnTerminal() throws IOException, TimeoutException, InterruptedException {
Address[] addresses = new Address[]{
new Address(PropertiesUtil.getProperty(RABBITMQ_HOST))
};
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(PropertiesUtil.getProperty(RABBITMQ_USER_NAME));
factory.setPassword(PropertiesUtil.getProperty(RABBITMQ_USER_PASSWORD));
Connection connection = factory.newConnection(addresses);
final Channel channel = connection.createChannel();
channel.basicQos(64); // 设置客户端最多接收未ack的消息个数，避免客户端被冲垮（常用于限流）
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body) throws IOException {
// 1.接收数据，并反序列化出对象
Terminal receiveTerminalConfig = JsonUtil.string2Obj(new String(body), Terminal.class);
// 2.验证是否是该终端的消息的消息 // 避免ACK其他终端的消息
if (receiveTerminalConfig.getMac() == GuavaCache.getKey(TERMINAL_MAC)) {
// 业务代码
ServerResponse response = iTerminalService.receiveTerminalFromRabbitmq(receiveTerminalConfig);
if (response.isSuccess()) {
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
}
};
channel.basicConsume(TERMINAL_CONFIG_CENTCONTROL2TERMINAL_QUEUE, consumer);
// 等回调函数执行完毕后，关闭资源
// 想了想还是不关闭资源，保持一个监听的状态，从而确保配置的实时更新
// TimeUnit.SECONDS.sleep(5);
// channel.close();
// connection.close();
}
}
小结：
这是早期写的一个demo代码，是直接参照源码的。如果是学习RabbitMq的话，还是建议手写一下这种比较原始的程序，了解其中每个方法的作用，从而理解RabbitMq的思路。如果条件允许的话，还可以查看一下RabbitMq的底层通信协议-AMQP（如果不方便下载，也可以私聊我）。
当然，此处可以通过@Value直接导入相关配置（乃至到了SpringCloud后，可以通过@Refreshscope等实现配置自动更新）。
与Spring集成：
producer:
package com.renewable.terminal.rabbitmq.producer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import com.renewable.terminal.pojo.Terminal;
import com.renewable.terminal.util.JsonUtil;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Description：
* @Author: jarry
*/
@Component("TerminalProducer")
public class TerminalProducer {
private static String rabbitmqHost = "47.92.249.250";
private static String rabbitmqUser = "admin";
private static String rabbitmqPassword = "123456";
private static String rabbitmqPort = "5672";
private static final String IP_ADDRESS = rabbitmqHost;
private static final int PORT = Integer.parseInt(rabbitmqPort);
private static final String USER_NAME = rabbitmqUser;
private static final String USER_PASSWORD = rabbitmqPassword;
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE = "exchange-terminal-config-terminal2centcontrol";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE = "queue-terminal-config-terminal2centcontrol";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINETYPE = "topic";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_BINDINGKEY = "terminal.config.terminal2centcontrol";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINGKEY = "terminal.config.terminal2centcontrol";
public static void sendTerminalConfig(Terminal terminal) throws IOException, TimeoutException, InterruptedException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(IP_ADDRESS);
factory.setPort(PORT);
factory.setUsername(USER_NAME);
factory.setPassword(USER_PASSWORD);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINETYPE, true, false, null);
channel.queueDeclare(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE, true, false, false, null);
channel.queueBind(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_BINDINGKEY);
String terminalStr = JsonUtil.obj2String(terminal);
channel.basicPublish(TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE, TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINGKEY, MessageProperties.PERSISTENT_TEXT_PLAIN, terminalStr.getBytes());
channel.close();
connection.close();
}
}
consumer:
package com.renewable.terminal.rabbitmq.consumer;
import com.rabbitmq.client.*;
import com.renewable.terminal.Init.SerialSensorInit;
import com.renewable.terminal.Init.TerminalInit;
import com.renewable.terminal.common.GuavaCache;
import com.renewable.terminal.common.ServerResponse;
import com.renewable.terminal.pojo.Terminal;
import com.renewable.terminal.service.ITerminalService;
import com.renewable.terminal.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import static com.renewable.terminal.common.constant.CacheConstant.TERMINAL_ID;
import static com.renewable.terminal.common.constant.CacheConstant.TERMINAL_MAC;
/**
* @Description：
* @Author: jarry
*/
@Component
@Slf4j
public class TerminalConsumer {
@Autowired
private ITerminalService iTerminalService;
@Autowired
private SerialSensorInit serialSensorInit;
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE = "exchange-terminal-config-centcontrol2terminal";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE = "queue-terminal-config-centcontrol2terminal";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINETYPE = "topic";
private static final String TERMINAL_CONFIG_TERMINAL2CENTCONTROL_BINDINGKEY = "terminal.config.centcontrol2terminal";
//TODO_FINISHED 2019.05.16 完成终端机TerminalConfig的接收与判断（ID是否为长随机数，是否需要重新分配）
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = TERMINAL_CONFIG_TERMINAL2CENTCONTROL_QUEUE, declare = "true"),
exchange = @Exchange(value = TERMINAL_CONFIG_TERMINAL2CENTCONTROL_EXCHANGE, declare = "true", type = TERMINAL_CONFIG_TERMINAL2CENTCONTROL_ROUTINETYPE),
key = TERMINAL_CONFIG_TERMINAL2CENTCONTROL_BINDINGKEY
))
@RabbitHandler
public void messageOnTerminal(@Payload String terminalStr, @Headers Map<String, Object> headers, Channel channel) throws IOException {
Terminal terminal = JsonUtil.string2Obj(terminalStr, Terminal.class);
if (terminal == null){
log.info("consume the null terminal config !");
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
channel.basicAck(deliveryTag, false);
}
if (!GuavaCache.getKey(TERMINAL_MAC).equals(terminal.getMac())){
log.info("refuse target terminal with mac({}) configure to this terminal with mac({}).",terminal.getMac(), GuavaCache.getKey(TERMINAL_MAC));
return;
}
// 2.业务逻辑
ServerResponse response = iTerminalService.receiveTerminalFromRabbitmq(terminal);
log.info("start serialSensorInit");
serialSensorInit.init();
// 3.确认
if (response.isSuccess()) {
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
channel.basicAck(deliveryTag, false);
}
}
}
配置：
# rabbitmq 消费端配置
spring:
rabbitmq:
listener:
simple:
concurrency: 5
max-concurrency: 10
acknowledge-mode: manual
# 限流
prefetch: 1
host: "localhost"
port: 5672
username: "admin"
password: "123456"
virtual-host: "/"
connection-timeout: 15000
小结：
这里不得不赞一下Spring，它通过提供RabbitMq地封装API-ampq，极大地简化了消息队列的代码。其实上述方法就是通过ampq的注解与yml配置来迅速实现RabbitMq的使用。
当然，这里还有很多的提升空间。比如说，通过@Bean注解（建立目标配置）与公用方法提取，可以有效提高代码复用性。
简单扩展（与SpringStream集成）：
这段代码并不是线上的代码，而是慕课网学习时留下的代码。主要实际生产中并没有使用SpringStream，但这确实是认识事件驱动模型的要给很好途径。
producer:
“`java
package com.imooc.order.message;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* @Description：
* @Author: jarry
*/
public interface StreamClient {
String INPUT = "myMessage";
String INPUT2 = "myMessageACK";
@Input(StreamClient.INPUT)
SubscribableChannel input();
@Output(StreamClient.INPUT)
MessageChannel output();
@Input(StreamClient.INPUT2)
SubscribableChannel input2();
@Output(StreamClient.INPUT2)
MessageChannel output2();
}“`
package com.imooc.order;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* @Description：
* @Author: jarry
*/
@Component
public class MqSenderTest extends OrderApplicationTests{
@Autowired
private AmqpTemplate amqpTemplate;
@Test
public void send(){
amqpTemplate.convertAndSend("myQueue", "now: " + new Date());
Assert.assertNotNull(new Date());
}
}
consumer:
package com.imooc.order.message;
import com.imooc.order.dto.OrderDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;
/**
* @Description：
* @Author: jarry
*/
@Component
@EnableBinding(StreamClient.class)
@Slf4j
public class StreamReceiver {
// @StreamListener(StreamClient.INPUT)
// public void process(Object message){
// log.info("StreamReceiver: {}", message);
// }
@StreamListener(StreamClient.INPUT)
// 增加以下注解，可以在INPUT消息消费后，返回一个消息。说白了就是RabbitMq对消息消费后的确认回调函数（貌似叫这个，意思就这样，之后细查）
@SendTo(StreamClient.INPUT2)
public String process(OrderDTO message){
log.info("StreamReceiver: {}", message);
return "received.";
}
@StreamListener(StreamClient.INPUT2)
public void process2(String message){
log.info("StreamReceiver2: {}", message);
}
}
总结：
在学习技术的过程中，一方面不断地感受到自己对技术了解的不足，另一方面则是发现更重要的是系统设计中技术选型的权衡。
相关文章