RabbitMQ消息队列安装及在Myeclipse下的开发实例
2017-05-05 21:58
381 查看
RabbitMQ安装教程
RabbitMQ是流行的开源消息队列系统,用erlang语言开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。
第一步:安装Erlang环境
下载地址:http://www.erlang.org/download
otp_win64_19.1.exe
下载exe执行文件直接安装,设置环境变量。
第二步:安装RabbitMQ
下载地址:http://www.rabbitmq.com/
rabbitmq-server-3.6.8.exe
下载exe执行文件直接安装即可。
执行命令:rabbitmq-plugins enable rabbitmq_management
开启网页支持 http://localhost:15672 默认用户:guest/guest
执行命令:rabbitmq-server start 开启服务
第三步:Myeclipse开发RabbitMQ实例—Topic Exchange模式
所需jar包在www.rabbitmq.com上下载的rabbitmq-java-client-bin-3.6.4
RabbitMQ提供了四种Exchange模式:fanout,direct,topic,header 。 本文使用topic作为示例。
所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上,Exchange 将RouteKey 和某Topic 进行模糊匹配。此时队列需要绑定一个Topic。可以使用通配符进行模糊匹配,符号“#”匹配一个或多个词,符号“”匹配不多不少一个词。因此“key.#”能够匹配到“key.1.2”,但是“key.” 只会匹配到“key.1”。所以,Topic Exchange 使用非常灵活。
生产者:
消费者1:
消费者2:
注:要引入rabbitmq-client.jar的包
代码结构如下:
先启动生产者进程产生消息,再启动消费者进程,可在网页上看到路由test_exchange_topic,有两个待消费的队列test_queue_topic_work1,test_queue_topic_work2
此时先自动生产者在启动两个消费者:可以看到消费者1接收不到消息,消费者2可以接收到消息。当把生产者的key.1.2改为key.1时可以发现两个消费者都能接收到消息。
RabbitMQ是流行的开源消息队列系统,用erlang语言开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。
第一步:安装Erlang环境
下载地址:http://www.erlang.org/download
otp_win64_19.1.exe
下载exe执行文件直接安装,设置环境变量。
第二步:安装RabbitMQ
下载地址:http://www.rabbitmq.com/
rabbitmq-server-3.6.8.exe
下载exe执行文件直接安装即可。
执行命令:rabbitmq-plugins enable rabbitmq_management
开启网页支持 http://localhost:15672 默认用户:guest/guest
执行命令:rabbitmq-server start 开启服务
第三步:Myeclipse开发RabbitMQ实例—Topic Exchange模式
所需jar包在www.rabbitmq.com上下载的rabbitmq-java-client-bin-3.6.4
RabbitMQ提供了四种Exchange模式:fanout,direct,topic,header 。 本文使用topic作为示例。
所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上,Exchange 将RouteKey 和某Topic 进行模糊匹配。此时队列需要绑定一个Topic。可以使用通配符进行模糊匹配,符号“#”匹配一个或多个词,符号“”匹配不多不少一个词。因此“key.#”能够匹配到“key.1.2”,但是“key.” 只会匹配到“key.1”。所以,Topic Exchange 使用非常灵活。
生产者:
package com; import com.rabbitmq.client.*; public class Sender { private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception{ //获取到连接以及mq通道 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); //factory.setPort(5672); //factory.setUsername("geust"); //factory.setPassword("geust"); Connection connection =factory.newConnection(); Channel channel = connection.createChannel(); // 声明exchange channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 消息内容 String message = "Hello World!"; channel.basicPublish(EXCHANGE_NAME, "key.1.2", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }
消费者1:
package com; import com.rabbitmq.client.*; public class RecvTest1 { private final static String QUEUE_NAME = "test_queue_topic_work1"; private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { System.out.println(" recive 1 start."); // 获取到连接以及mq通道 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); //factory.setPort(5672); //factory.setUsername("geust"); //factory.setPassword("geust"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.*"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
消费者2:
package comm; import com.rabbitmq.client.*; public class RecvTest2 { private final static String QUEUE_NAME = "test_queue_topic_work2"; private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { System.out.println(" recive 2 start."); // 获取到连接以及mq通道 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); //factory.setPort(5672); //factory.setUsername("geust"); //factory.setPassword("geust"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.#"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
注:要引入rabbitmq-client.jar的包
代码结构如下:
先启动生产者进程产生消息,再启动消费者进程,可在网页上看到路由test_exchange_topic,有两个待消费的队列test_queue_topic_work1,test_queue_topic_work2
此时先自动生产者在启动两个消费者:可以看到消费者1接收不到消息,消费者2可以接收到消息。当把生产者的key.1.2改为key.1时可以发现两个消费者都能接收到消息。
至此,RabbitMQ消息队列安装及在Myeclipse下的开发完成,相比UNIX的IPC的单机局限性,RabbitMQ的接口简单及在分布式上应用很大。
相关文章推荐
- ActiveMQ开发实例4 查看队列消息
- RabbitMq 消息队列的安装、使用 、原理
- RabbitMQ消息队列入门篇(环境配置+Java实例+基础概念)
- RabbitMQ消息队列(2)安装RabbitMQ
- java调用rabbitmq消息队列发送和接收消息实例
- debian(wheezy)安装rabbitMQ,php-amqp来支持消息队列.
- centos下安装RabbitMQ消息队列
- Spring整合RabbitMQ进行消息队列开发
- RabbitMQ消息队列入门篇(环境配置+Java实例+基础概念)
- 使用专业的消息队列产品rabbitmq之centos7环境安装
- openstack icehouse 安装消息队列模块rabbitmq 推荐
- RabbitMQ实例教程:RPC远程调用消息队列
- RabbitMQ消息队列安装和配置以及推送消息
- php 消息队列 rabbitmq 的安装使用
- Centos5.4 消息队列rabbitmq的安装及使用
- linux下安装配置rabbitmq(消息队列系统)
- 柯南君:看大数据时代下的IT架构(3)消息队列之RabbitMQ-安装、配置与监控
- RabbitMQ实例教程:发布/订阅者消息队列
- 消息队列RabbitMQ的安装
- centos安装rabbitmq消息队列