RabbitMQ(二) - HelloWorld

Hello World

这部分的教程我们将写两个程序:一个生产者发送一条消息和一个消费者接收消息并打印出来。我们会略过java api的一些细节的地方,我们从非常简单的开始。一个Hello World的消息。

在下图总,P是我们的生产者,C是我们的消费者。中间的箱子是队列 —— 代表消费者的消息缓存区。

rabbitmq-helloword

The Java client library

RabbitMQ支持多种协议。这个教程使用AMQP 0-9-1,开源的、通用的消息协议。有许多不同的语言实现RabbitMQ,此处我们用java来实现。

下载客户端库,它依赖SLF4J APISLF4J Simple。拷贝这些文件到你的工作目录。

请注意:SLF4J Simple对这个教程来说已经足够,但是最好使用完全成熟的日志开发库,例如:logback。

RabbitMQ的java客户端也支持Maven,groupId是com.rabbitmq和artifactId是amqp-client

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>4.0.2</version>
    <scope>compile</scope>
</dependency>

现在我们有了java的客户端以及它的依赖,我们可以写一些代码了。

NOTE:官网是没有使用IDE的。这里我用的是idea,毕竟大家都是开发者,IDE会方便点。

发送(Sending)

rabbitmq-helloworld-sending

我们将调用我们的消息生产者(发送者)Send和消息的消费者(接收者)Recv。生产者将连接到RabbitMQ,发送一条消息,然后退出。

Send.java中,我们需要导入一些类:

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

设置类和队列名:

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

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

然后我们创建一个服务连接:

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

这个连接抽象了socket连接,维护版本协议和身份认证。这里我们用的是本地机器——因此是localhost。如果我们想在不同的机器连接到中间件,只需要指定不同的名字和IP地址即可。

接下来我们创建一个通道(channel)。

我们还需要定义一个队列供我们将消息推送给它。

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 + "'");

定义一个队列是幂等的——它只有在队列不存在的时候创建。消息的内容是字节数组,所以你能编码成任何你想要的数据。

最后,我们要关闭通道(channel)和连接(conection);

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

下面是完整的类:Send.java

package com.roachfu.tutorial.rabbitmq.website.helloworld;

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 {

    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        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();

    }
}

接收端(Receiving)

消费者从RabbitMQ中拉取消息,不同于生产者生产一条消息,我们会一直监听消息并答应它们。

rabbitmq-helloworld-receiving

代码(在Recv.java中)导入的和Send导入的差不多:

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;

额外不同的DefaultConsumer是一个实现了Consumer接口的类,通过服务拉取缓存的消息。

设置也和publisher差不多,我们打开一个连接和通道并定义一个队列,需要注意的是队列名和Send.java中定义的一样。

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, false, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    ...
    }
}

注意:在这里我们也要定义队列,因为我们有可能在生产者之前启动消费者,我们需要保证我们在消费消息的时候队列是存在的。

我们要告知服务从队列中将消息交付给我们。由于它将异步的推送消息给我们,我们提供一个回调对象缓存消息直到我们已经使用了它们。这就是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);

以下是Recv.java完整代码

package com.roachfu.tutorial.rabbitmq.website.helloworld;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 消费者
 */
public class Recv {

    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {

        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. . . ");

        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);
    }
}

输出

  • 生产者

[x] Sent 'Hello World'

  • 消费者

[*] Waiting for messages. . .
[x] Received 'Hello World'

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

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,717评论 18 139
  • 来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,369评论 2 34
  • 关于消息队列,从前年开始断断续续看了些资料,想写很久了,但一直没腾出空,近来分别碰到几个朋友聊这块的技术选型,是时...
    预流阅读 584,972评论 51 786
  • 他们都说,讲一个故事要讲时间,地点,人物,事件。然后按起始,发展,高潮,结尾,慢慢的讲。 我可不会。 或许我也会的...
    4d8ef48cfd78阅读 290评论 0 0
  • 我总是一个人走在路上 任何时候都不会忘记张望 陌生人在身旁 也不会害怕随波流浪 我带着自己的行囊 重到从不会迷失而...
    敬言Wzing阅读 284评论 2 3