RabbitMQ：快速入门hello word

2018-06-21 本文已影响112人 AubreyXue

封面.jpg

1、相关概念

RabbitMQ是一个消息代理，事实上，它接收生产者产生的消息，然后将消息传递给消费者。在这个过程中，它可以路由，可以缓冲，或者更具你设定的规则来将消息持久化。RabbitMQ和消息传输过程中一般会用一些术语：

生产者（Producing）：

意思无非是指发送消息的那一端，如果一个程序发送消息，那么它就将被称为生产者，这里用大写的P来表示。

image

队列（queue）：

相当于邮箱的名字，它活动在RabbitMQ服务器里边。虽然消息流会通过RabbitMQ和你的应用程序，但是只会被存储在队列中。队列是不受任何限制的，它可以尽可能多的去存储你需要存储的消息（本质上来说它是个无限缓冲）。可以多个生产者向同一个消息队列发送消息，也可以多个消费者同时从一个消息队列中来接收消息。消息队列可以如下图模型。

image

消费者（Consuming）：

即接收端，消费者主要是等待接收消息的程序，用下图表示:

image

注意：在大多数应用场景中，生产者、消费者以及RabbitMQ服务是不会同时运行在一台机器上的。

下边将会实现两个Java程序，一个只发送生产者一条消息的生产者，一个接收消息、并打印消息的消费者。

在下边的对话中，”P”是我们的生产者，”C”是我们的消费者，中间的是矩形是队列（BabbitMQ维护的消息缓冲）

image

2、pom引入RabbitMQ的包

<dependency>
     <groupId>com.rabbitmq</groupId>
     <artifactId>amqp-client</artifactId>
     <version>4.1.0</version>
</dependency>

3、生产者（发送端）

image.png

我们将消息发送器起名为Send，消息接收器起名为Recv。发送器将会连接RabbitMQ，发送一条消息，然后退出。

package com.hrabbit.rabbitmq.send;

import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * @Auther: hrabbit
 * @Date: 2018-06-21 上午10:01
 * @Description:
 */
public class Send {

    private static final String QUEUE_NAME="queue_hrabbit";

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

        String message="hello word!";
        //获取连接
        Connection connection = ConnectionUtils.getConnection();
        //创建通道，为了发送消息，你必须要声明一个消息消息队列，然后向该队列里推送消息
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //发送消息
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());

        channel.close();

        connection.close();
    }
}

4、消费者（接收端）

RabbitMQ会往接收器上推消息，与只发送一条消息的发送端不同，这里我们将监听消息并将消息打印出来。

image

package com.hrabbit.rabbitmq.recover;

import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;

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

/**
 * @Auther: hrabbit
 * @Date: 2018-06-21 上午10:45
 * @Description:
 */
public class Recover {

    private static final String QUEUE_NAME="queue_hrabbit";

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

        //获取connection连接
        Connection connection = ConnectionUtils.getConnection();
        //创建通道，你必须要声明一个消息消息队列，然后向该队列里推送消息
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //在低版本中，new QueueingConsumer();的方式，但是这种方式已经被废弃了，不建议使用
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, "utf-8");
                System.out.println("msg:"+msg);
            }
        };

        channel.basicConsume(QUEUE_NAME,defaultConsumer);
    }
}

5、工具类（获取Connection）

package com.hrabbit.rabbitmq.utils;

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

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

/**
 * @Auther: hrabbit
 * @Date: 2018-06-21 上午9:58
 * @Description:
 */
public class ConnectionUtils {

    /**
     * 获取connection
     * @return
     * @throws IOException
     * @throws TimeoutException
     */
    public static Connection getConnection() throws IOException, TimeoutException {
        //创建connection工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置ip
        factory.setHost("127.0.0.1");
        //设置端口号
        factory.setPort(5672);
        //设置用户名称
        factory.setUsername("hrabbit");
        //设置密码
        factory.setPassword("123");
        //设置vhost
        factory.setVirtualHost("/hrabbit");
        return factory.newConnection();
    }
}