消息队列实践(一)——RabbitMQ遇到docker

本文尝试将docker作为管理消息队列的服务器,消息队列的producer和comsumer部署在本机。为达到此目的,会用两种方式实现:

  • RabbitMQ:官网推荐方式,队列由RabbitMQ-cli进行管理;
  • RabbitMQ-Management:这种方式是为用户提供图形化的管理界面,进行队列任务及状态的管理;


    基于docker的MQ实践

按说事先应阐明什么是消息队列,什么样的场景需要有消息队列。但本文想着重于代码层面的实现,所以关于这个问题,可能会另辟一文来详细解释。如果你看过RabbitMQ官网的 helloworld篇,文中是将send、receive和queue分成三个服务进行演示。作者为了方便理解,将send和receive放在同一个codebase,只是通过不同的Restapi控制而已。我们安装官网Tutorials推荐的方式进行介绍:

  • docker的配置

  1. 本机docker的安装配置本文就不赘述。我们可以在RabbitMQ官网的Download中找到下载镜像的地址
  2. 将镜像拉到本地docker:docker pull rabbitmq
  3. 运行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,这里推荐一篇很清晰的博文,就不赘述了。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 214,951评论 6 497
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 91,606评论 3 389
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 160,601评论 0 350
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 57,478评论 1 288
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 66,565评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 50,587评论 1 293
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,590评论 3 414
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 38,337评论 0 270
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,785评论 1 307
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,096评论 2 330
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,273评论 1 344
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,935评论 5 339
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,578评论 3 322
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,199评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,440评论 1 268
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 47,163评论 2 366
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,133评论 2 352

推荐阅读更多精彩内容

  • http://liuxing.info/2017/06/30/Spring%20AMQP%E4%B8%AD%E6%...
    sherlock_6981阅读 15,904评论 2 11
  • 1.RabbitMQ概述 简介: MQ全称为Message Queue,消息队列是应用程序和应用程序之间的通信方法...
    梁朋举阅读 49,580评论 0 47
  • 什么叫消息队列 消息(Message)是指在应用间传送的数据。消息可以非常简单,比如只包含文本字符串,也可以更复杂...
    lijun_m阅读 1,343评论 0 1
  • 关于消息队列,从前年开始断断续续看了些资料,想写很久了,但一直没腾出空,近来分别碰到几个朋友聊这块的技术选型,是时...
    预流阅读 584,623评论 51 786
  • 关于消息队列,从前年开始断断续续看了些资料,想写很久了,但一直没腾出空,近来分别碰到几个朋友聊这块的技术选型,是时...
    Johnson_zx阅读 1,108评论 0 5