本文尝试将docker作为管理消息队列的服务器,消息队列的producer和comsumer部署在本机。为达到此目的,会用两种方式实现:
- RabbitMQ:官网推荐方式,队列由RabbitMQ-cli进行管理;
-
RabbitMQ-Management:这种方式是为用户提供图形化的管理界面,进行队列任务及状态的管理;
基于docker的MQ实践
按说事先应阐明什么是消息队列,什么样的场景需要有消息队列。但本文想着重于代码层面的实现,所以关于这个问题,可能会另辟一文来详细解释。如果你看过RabbitMQ官网的 helloworld篇,文中是将send、receive和queue分成三个服务进行演示。作者为了方便理解,将send和receive放在同一个codebase,只是通过不同的Restapi控制而已。我们安装官网Tutorials推荐的方式进行介绍:
-
docker的配置
- 本机docker的安装配置本文就不赘述。我们可以在RabbitMQ官网的Download中找到下载镜像的地址;
- 将镜像拉到本地docker:
docker pull rabbitmq
- 运行image:
docker run -p 5672:5672 -d --hostname my-rabbit --name some-rabbit rabbitmq:3
这里会做一个端口映射,因为5672是RabbitMQ的默认运行端口,也就是说在docker中MQ的端口号为5672,当我们在Springboot项目中引入RabbitMQ库后,如果没有特别配置,Springboot会默认和指定host的5672端口进行连接,所以我们将本地5672和docker5672打通,这样后续就无需做端口上的配置。
到这里我们docker中部署RabbitMQ就完成了,可以通过以下命令来看一下部署情况:
docker ps——查看正在运行的docker container
docker ps some-rabbit——some-rabbit的端口映射
docker exec -it 2758fde2d00c bash——进入2758fde2d00c容器
rabbitmqctl list_queues——显示所有队列和各自message数(在我的队列列表中已经存在两个队列)
队列最基本的组成不难想象:消息发送方、队列容器、消息接收方。在RabbitMQ中有专门的称谓: Publishers 、broker、Consumers。各种名词解释参见RabbitMQ Document(Client Libraries and Features)
-
Send
先用Spring Initializr快速构建项目,选择Spring Web Starter和Spring Rabbitmq
创建Send.class如下所示:
package com.otof.rabbitmq.send;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Send {
public static final String QUEUE_NAME = "MYQUEUE";
public void sendMessage(String message) {
ConnectionFactory factory = new ConnectionFactory();
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "'");
} catch (TimeoutException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
}
我们一行行来分析,在方法sendMessage中,首先创建一个ConnectionFactory实例,然后通过factory.newConnection(),创建connection,如果我们点进去看源码的话
public Connection newConnection() throws IOException, TimeoutException {
return this.newConnection(this.sharedExecutor, Collections.singletonList(new Address(this.getHost(), this.getPort())));
}
可以发现newConnection在不断调用重载方法,而最外层传了两个参数:sharedExecutor和host/port,可以发现如果我们不去设置host和port的时候,它会自动初始化为localhost/5672。这里有一个不常见的try-catch-resource方式,因为建立producer和broker的连接是要耗费资源的,就如同我们建立数据库连接一样,所以一旦基于该连接的业务执行完成,我们应该释放资源,try-catch-resource是jdk1.7提出的新写法,它意味任何实现了java.lang.AutoCloseable和java.io.Closeable的对象,都可以在语句执行之后,自动关闭资源。很明显的一个例子就是我们常用的读取文件操作:
public class PreJDK7 {
public static String readFirstLingFromFile(String path) throws IOException {
BufferedReader br = null;
try {
br = new BufferedReader(new FileReader(path));
return br.readLine();
} catch (IOException e) {
e.printStackTrace();
} finally {//必须在这里关闭资源
if (br != null)
br.close();
}
return null;
}
}
而使用try-catch-resource:
public class AboveJDK7 {
static String readFirstLineFromFile(String path) throws IOException {
try (BufferedReader br = new BufferedReader(new FileReader(path))) {
return br.readLine();
}
}
}
接着回到Send.class在创建连接之后,会创建一个channel:
Channel channel = connection.createChannel()
它依然会在使用完成后自动关闭,channel会声明队列,而后发布该队列。这样一个简单的Send Demo就完成了。这里有两个概念:connection
和channel,可以这样说connection是连接producer和broker的实际管道,而channel是依附于这个管道的,多根连接线,所以实际的信息传输是通过channel完成的
connections are multiplexed with channels that can be thought of as "lightweight connections that share a single TCP connection".
channel.queueDeclare的传参是用来配置该队列的基本属性
queueDeclare(String queue,
boolean durable,
boolean exclusive,
Map<String, Object> arguments);
queue: 队列名称;
durable: 是否持久化, 队列的声明默认是存放到内存中的,如果rabbitmq重启会丢失,如果想重启之后还存在就要使队列持久化,保存到Erlang自带的Mnesia数据库中,当rabbitmq重启之后会读取该数据库;
exclusive:是否排外的,有两个作用,一:当连接关闭时connection.close()该队列是否会自动删除;二:该队列是否是私有的private,如果不是排外的,可以使用两个消费者都访问同一个队列,没有任何问题,如果是排外的,会对当前队列加锁,其他通道channel是不能访问的
arguments: 队列中的消息什么时候会自动被删除?(此处可自行翻阅API)
channel.basicPublish可以先理解为发布这个queue到broker中,关于发布的话题比较复杂,可以再找机会讨论。
以上就是Send的相关解析,类似的,Recv或者叫consumer也是这样的流程:
package com.otof.rabbitmq.receive;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Recv {
private final String QUEUE_NAME = "MYQUEUE";
public void receiveMessage() throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("localhost");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
};
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
}
我们可以通过Restful API来触发Send和Recv
@RestController
public class RabbitmqController {
@PostMapping(path = "addToQueue")
public void addMessageToQueue(@RequestBody String message) {
new Send().sendMessage(message);
}
@GetMapping(path = "getFromQueue")
public void getMessageFromQueue() throws IOException, TimeoutException {
new Recv().receiveMessage();
}
}
当请求addToQueue后,我们可以进入docker看一下当前queue_list,会发现新建的MYQUEUE已经进入broker
queue list
然后请求get方法,拿取MYQUEUE当中的内容,在console中可以看到打印的日志,代表从broker中获得了队列信息。
get
console log
此时我们再看一眼docker中的队列列表,会发现已经为0
queue list
这就是本地使用docker实现RabbitMQ的demo,还有一种docker管理的方式,就是使用RabbitMQ-Management,这里推荐一篇很清晰的博文,就不赘述了。
网友评论