RabbitMQ消息队列安装及在Myeclipse下的开发实例

RabbitMQ安装教程
RabbitMQ是流行的开源消息队列系统&＃xff0c;用erlang语言开发。RabbitMQ是AMQP&＃xff08;高级消息队列协议&＃xff09;的标准实现。

第一步&＃xff1a;安装Erlang环境
下载地址:http://www.erlang.org/download
otp_win64_19.1.exe
下载exe执行文件直接安装&＃xff0c;设置环境变量。

第二步&＃xff1a;安装RabbitMQ
下载地址:http://www.rabbitmq.com/
rabbitmq-server-3.6.8.exe
下载exe执行文件直接安装即可。

执行命令&＃xff1a;rabbitmq-plugins enable rabbitmq_management
开启网页支持 http://localhost:15672 默认用户&＃xff1a;guest/guest
执行命令&＃xff1a;rabbitmq-server start 开启服务

第三步&＃xff1a;Myeclipse开发RabbitMQ实例—Topic Exchange模式
所需jar包在www.rabbitmq.com上下载的rabbitmq-java-client-bin-3.6.4

　　RabbitMQ提供了四种Exchange模式&＃xff1a;fanout,direct,topic,header 。 本文使用topic作为示例。
　
所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上&＃xff0c;Exchange 将RouteKey 和某Topic 进行模糊匹配。此时队列需要绑定一个Topic。可以使用通配符进行模糊匹配&＃xff0c;符号“#”匹配一个或多个词&＃xff0c;符号“”匹配不多不少一个词。因此“key.#”能够匹配到“key.1.2”&＃xff0c;但是“key.” 只会匹配到“key.1”。所以&＃xff0c;Topic Exchange 使用非常灵活。

生产者&＃xff1a;

package com;import com.rabbitmq.client.*;public class Sender {private final static String EXCHANGE_NAME &＃61; "test_exchange_topic";public static void main(String[] argv) throws Exception{//获取到连接以及mq通道ConnectionFactory factory &＃61; new ConnectionFactory();factory.setHost("127.0.0.1");//factory.setPort(5672);//factory.setUsername("geust");//factory.setPassword("geust");Connection connection &＃61;factory.newConnection();Channel channel &＃61; connection.createChannel(); // 声明exchange channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 消息内容 String message &＃61; "Hello World!"; channel.basicPublish(EXCHANGE_NAME, "key.1.2", null, message.getBytes()); System.out.println(" [x] Sent &＃39;" &＃43; message &＃43; "&＃39;"); channel.close(); connection.close();}
}

消费者1&＃xff1a;

package com;
import com.rabbitmq.client.*;public class RecvTest1 {private final static String QUEUE_NAME &＃61; "test_queue_topic_work1"; private final static String EXCHANGE_NAME &＃61; "test_exchange_topic"; public static void main(String[] argv) throws Exception { System.out.println(" recive 1 start.");// 获取到连接以及mq通道 ConnectionFactory factory &＃61; new ConnectionFactory();factory.setHost("127.0.0.1");//factory.setPort(5672);//factory.setUsername("geust");//factory.setPassword("geust");Connection connection &＃61; factory.newConnection(); Channel channel &＃61; connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.*"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer &＃61; new QueueingConsumer(channel); // 监听队列&＃xff0c;手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery &＃61; consumer.nextDelivery(); String message &＃61; new String(delivery.getBody()); System.out.println(" [x] Received &＃39;" &＃43; message &＃43; "&＃39;"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } }
}

消费者2&＃xff1a;

package comm;
import com.rabbitmq.client.*;public class RecvTest2 {private final static String QUEUE_NAME &＃61; "test_queue_topic_work2"; private final static String EXCHANGE_NAME &＃61; "test_exchange_topic"; public static void main(String[] argv) throws Exception { System.out.println(" recive 2 start.");// 获取到连接以及mq通道 ConnectionFactory factory &＃61; new ConnectionFactory();factory.setHost("127.0.0.1");//factory.setPort(5672);//factory.setUsername("geust");//factory.setPassword("geust"); Connection connection &＃61; factory.newConnection(); Channel channel &＃61; connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.#"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer &＃61; new QueueingConsumer(channel); // 监听队列&＃xff0c;手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery &＃61; consumer.nextDelivery(); String message &＃61; new String(delivery.getBody()); System.out.println(" [x] Received &＃39;" &＃43; message &＃43; "&＃39;"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } }
}

注&＃xff1a;要引入rabbitmq-client.jar的包
代码结构如下&＃xff1a;

先启动生产者进程产生消息&＃xff0c;再启动消费者进程&＃xff0c;可在网页上看到路由test_exchange_topic&＃xff0c;有两个待消费的队列test_queue_topic_work1&＃xff0c;test_queue_topic_work2


此时先自动生产者在启动两个消费者&＃xff1a;可以看到消费者1接收不到消息&＃xff0c;消费者2可以接收到消息。当把生产者的key.1.2改为key.1时可以发现两个消费者都能接收到消息。

至此&＃xff0c;RabbitMQ消息队列安装及在Myeclipse下的开发完成&＃xff0c;相比UNIX的IPC的单机局限性&＃xff0c;RabbitMQ的接口简单及在分布式上应用很大。