RabbitMQ初探-JAVA

RabbitMQ是一个消息队列中间件.适用于生产者与消费者的情况。

最近因为融入新公司接触到这个玩意儿，于是想试用一下.因此把server搭建在本机环境

一.windows下的安装

1.http://www.rabbitmq.com/download.html下载windows版本(本机64位系统)

2.下载RabbitMQ所需环境erlang

http://www.erlang.org/downloads

（选择OTP 18.3 Windows 64-bit Binary File）

3.安装erlang并配置环境变量

安装完后新建一个环境变量命名为：ERLANG_HOME，值为安装的根目录

4.安装RabbitMQ

二.启动RabbitMQ server服务

cmd模式下进入RabbitMQ目录sbin,输入命令:

rabbitmq-server start

不过在此之前.我们需要先启动插件用于我们可以使用浏览器登录管理控制台界面.因此先中断服务.然后输入命令：

rabbitmq-plugins enable rabbitmq_management (启动插件)

之后再启动RabbitMQ服务即可.

三.简单的Demo

1.下载JAVA客户端jar并加入项目中

amqp-client-3.6.0.jar

2.生产者(发送消息方)

/**
* 队列生产者
* @author mingchenchen
*
*/
public class Producer {
public static void produce() throws IOException, TimeoutException{
/**
* 创建连接连接到MabbitMQ
*/
ConnectionFactory factory = new ConnectionFactory();
//设置MabbitMQ所在主机ip或者主机名
factory.setHost("127.0.0.1");
//创建一个连接
Connection connection = factory.newConnection();
//创建一个频道
Channel channel = connection.createChannel();
//指定一个队列
channel.queueDeclare(Main.QUEUE_NAME, false, false, false, null);
//发送的消息
String message = "hello world!";
//往队列中发出一条消息
channel.basicPublish("", Main.QUEUE_NAME, null, message.getBytes());
System.out.println("Sent '" + message + "'");
//关闭频道和连接
channel.close();
connection.close();
}
}

3.消费者（接收消息方）

/**
* 队列消费者
* @author mingchenchen
*
*/
public class Consumer {

public static void cost() throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException{
//打开连接和创建频道，与发送端一样
ConnectionFactory factory = new ConnectionFactory();
//设置MabbitMQ所在主机ip或者主机名
factory.setHost("127.0.0.1");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
channel.queueDeclare(Main.QUEUE_NAME, false, false, false, null);
System.out.println("Waiting for messages. To exit press CTRL+C");

//创建队列消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
//指定消费队列
channel.basicConsume(Main.QUEUE_NAME, true, consumer);
while (true)
{
//nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println("Received '" + message + "'");
}

}

}

3.

/**
* 主方法 设置生产者及消费者 并且执行各自的任务
* @author mingchenchen
*
*/
public class Main {
public static final String QUEUE_NAME = "QUEUE_TEST";

public static void main(String[] args) throws ShutdownSignalException, ConsumerCancelledException, InterruptedException {
try {
Producer.produce("RabbitMQ测试——————发送一条消息");

Thread.sleep(2000);

Consumer.cost();
} catch (IOException | TimeoutException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

结果:



可以看到，程序依然在运行，因为消费者默认是一直等待接受消息的。