消息队列实践（一）——RabbitMQ遇到docker

2019-07-21 本文已影响213人瑞瑞余之

本文尝试将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中MQ相关配置
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

Spring Initializr启动项目
创建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

post

queue list
然后请求get方法，拿取MYQUEUE当中的内容，在console中可以看到打印的日志，代表从broker中获得了队列信息。

get

console log
此时我们再看一眼docker中的队列列表，会发现已经为0

queue list
这就是本地使用docker实现RabbitMQ的demo，还有一种docker管理的方式，就是使用RabbitMQ-Management，这里推荐一篇很清晰的博文，就不赘述了。

消息队列实践（一）——RabbitMQ遇到docker

docker的配置

Send

猜你喜欢

热点阅读