【译】RabbitMQ教程一

内容来自:RabbitMQ Tutorials Java版


介绍

RabbitMQ是一个消息代理:它接受并转发消息。你可以把它当成一个邮局:当你想邮寄信件的时候,你会把信件放在投递箱中,并确信邮递员最终会将信件送到收件人的手里。在这个例子中,RabbitMQ就相当与投递箱、邮局和邮递员。

RabbitMQ与邮局的区别在于:RabbitMQ并不处理纸质信件,而是接受、存储并转发二进制数据---消息。

谈到RabbitMQ的消息,通常有几个术语:

  • 生产者:是指发送消息的程序
  • 队列:相当于RabbitMQ的投递箱。尽管消息在RabbitMQ和你的应用之间传递,但是消息仅仅会在队列之中存储。队列只能存储在内存或磁盘中,本质上是一个大的消息缓冲区。不同的生产者可以发送消息到同一个对队列,不同的消费者也可以从同一个队列中获取消息。
  • 消费者:等待接受消息的程序。

注意,生产者、消费者以及RabbitMQ并不一定要在同一个主机上,在绝大部分的应用中它们都不在同一主机上。

在开始教程之前,请确保:你已经安装了RabbitMQ,并且在localhost上运行起来(默认端口5672)。如果你使用了不同的主机或端口,请在下文中的连接设置中
更改相应的参数。


一、Hello World

在这一部分,我们将会使用Java编写两个小程序:一个发送单个消息的生产者、一个接受消息并打印出消息的消费者。这个消息就是Hello World

下图中,P代表生产者,C代表消费者,中间红色的小箱子就代表队列--RabbitMQ为了让消费者收到消息而保持的消息缓冲区。


Rabbit.png

在这一部分,只需要引入Java客户端依赖即可:amqp-client.jar,也可以通过maven的方式引入:

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>4.1.0</version>
</dependency>
1、生产者

我们将消息的发布者(生产者)命名为Send,将消息的消费者命名为Recv。发布者将会连接到RabbitMQ,并且发送一条消息,然后退出。
Send.java中,首先引入相关类:

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

再定义队列的名字:

private final static String QUEUE_NAME = "hello";

然后,创建一个连接到Rabbit服务器的连接:

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

上面的代码中,connection是socket连接的抽象,为我们处理了通信协议版本协商以及认证等。这样,我们就连接到了本地机器上的一个消息代理(broker)。如果想连接到其他机器上的broker,只要修改IP即可。

之后,我们又创建了一个通道(channel),大部分的API操作均在这里完成。

对于Send来说,必须指明消息要发到哪个队列:

channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");

队列的定义是幂等的,它仅仅在不存在时才会创建。消息的内容是一个字节数组,所以你可以随意编码(encode)。

最后,必须将通道和连接关闭。

channel.close();
connection.close();

完整代码

//引入相关Class文件
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Send {

    //定义队列名字
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        //创建连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //为通道指明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "Hello World!";
        
        //发布消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");

        //关闭连接
        channel.close();
        connection.close();
    }
}
2、接收者(消费者)

消费者从RabbitMQ中取出消息。不同于发布者只发送一条消息就退出,这里我们让消费者一直监听消息,并把接受到的消息打印出来。
与Send.java类似,首先引入相关类:

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;

上面引入的DefaultConsumerConsumer接口的实现类,我们使用它来缓冲从服务器push来的消息。
接下来的设置与发布者类似,打开连接和通道,声明我们想消费的队列。注意,这里的队列的名字要与发布者中声明的队列的名字一致。

public class Recv {
  private final static String QUEUE_NAME = "hello";

  public static void main(String[] argv)
      throws java.io.IOException,
             java.lang.InterruptedException {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(QUEUE_NAME, fasle, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    ...
    }
}

注意,消费者同样声明了队列。这是因为,我们可能在启动生产者之前启动了消费者应用,我们想确保在从一个队列消费消息之前,这个队列是存在的。

接下来,告诉服务器(RabbitMQ)把队列中的消息发过来。因为这个过程是异步的,可以通过DefaultConsumer来进行回调。

Consumer consumer = new DefaultConsumer(channel) {
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope,
                             AMQP.BasicProperties properties, byte[] body)
      throws IOException {
    String message = new String(body, "UTF-8");
    System.out.println(" [x] Received '" + message + "'");
  }
};
channel.basicConsume(QUEUE_NAME, true, consumer);

Consumer的完整代码如下:

package com.maxwell.rabbitdemo;

import com.rabbitmq.client.*;

import java.io.IOException;

public class Recv {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        //建立连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //声明要消费的队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        //回调消费消息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

这样,消费者就会一直监听声明的队列。运行一次生产者(即Send.java中的main方法),消费者就会打印出接受到的消息。


说明

①与原文略有出入,如有疑问,请参考原文。
②RabbitMQ的官方rabbitmq-tutorials的java示例中,amqp-client版本为3.5,我改为了4.1,否则后续的示例教程中会报错说找不到文件。

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

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,596评论 18 139
  • 来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,343评论 2 34
  • 关于消息队列,从前年开始断断续续看了些资料,想写很久了,但一直没腾出空,近来分别碰到几个朋友聊这块的技术选型,是时...
    预流阅读 584,386评论 51 785
  • 有些过程确实很煎熬,但真的,熬过去就好了,而且,我可以!
    小兔纸的胡萝卜阅读 95评论 1 1
  • 有时候觉得自己就是太二了,很多自己不愿的事但又觉得无所谓╮(╯_╰)╭的,就忍忍将就将就一下就可以了,可是这样真的...
    子絔627阅读 154评论 0 0