标签：
rabbitMQ是一种高性能的消息队列，支持或者说它实现了AMQP协议（advanced message queue protocol高级消息队列协议）。
下面简单讲一讲一个小例子。我们首先要部署好rabbitMQ，然后实现一个生产者—消费者，生产者向rabbit中发布一个消息，消费者去rabbit取这个消息，在正确收到这个消息后，消费者会通过返回队列回写通知生产者自己收到了消息。
windows下部署rabbit非常简单，先安装erlang运行时，然后安装rabbitMQ安装文件即可，都是exe的，很简单。然后找到rabbit的sbin目录里的bat即可启动rabbitMQ。
下面是producer—consumer代码：
package com.hzfi.rabbitmq; import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.QueueingConsumer.Delivery; import com.rabbitmq.client.ShutdownSignalException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.ConsumerCancelledException; import com.rabbitmq.client.QueueingConsumer; public class Producer { private final static String QUEUE_NAME = "myQueue"; //上送队列 public static void main(String[] args) throws IOException, TimeoutException{ String replyQueueName = null; //返回队列名 ConnectionFactory connFactory = null; Connection conn = null; Channel channel = null; try{ connFactory = new ConnectionFactory(); connFactory.setHost("localhost"); conn = connFactory.newConnection(); channel = conn.createChannel(); //返回队列 replyQueueName = channel.queueDeclare().getQueue(); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(replyQueueName, true, consumer); String corrId = java.util.UUID.randomUUID().toString(); //用来表示返回队列结果的id,唯一 BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build(); String msg = "linyang@hzfi.cn"; channel.queueDeclare(QUEUE_NAME, false, false, false, null); channel.basicPublish("", QUEUE_NAME, props, msg.getBytes()); System.out.println("producer has published: \"" + msg + "\""); while(true){ Thread.sleep(1000); Delivery delivery = consumer.nextDelivery(); System.out.println("from server reply:" + new String(delivery.getBody())); } }catch(IOException ioe){ ioe.printStackTrace(); }catch(TimeoutException toe){ toe.printStackTrace(); } catch (ShutdownSignalException e) { e.printStackTrace(); } catch (ConsumerCancelledException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); }finally{ if(channel!=null) channel.close(); if(conn!=null) conn.close(); } } }
package com.hzfi.rabbitmq; import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.client.QueueingConsumer.Delivery; public class Consumer { private final static String QUEUE_NAME = "myQueue"; public static void main(String[] args) throws IOException, TimeoutException{ ConnectionFactory connFactory = null; Connection conn = null; Channel channel = null; try{ connFactory = new ConnectionFactory(); connFactory.setHost("localhost"); conn = connFactory.newConnection(); channel = conn.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println("listening for event message..."); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(QUEUE_NAME, true, consumer); while(true){ Thread.sleep(1000); Delivery delivery = consumer.nextDelivery(); BasicProperties props = delivery.getProperties(); BasicProperties reply_props = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build(); String msg = new String(delivery.getBody(),"utf-8"); System.out.println("receive msg:" + msg); String retMsg = "ok, give you reply：" + new String(msg.getBytes(),"utf-8"); System.out.println("Consumer中的返回队列名" + props.getReplyTo()); channel.basicPublish( "", props.getReplyTo(), reply_props, retMsg.getBytes()); } }catch(IOException ioe){ ioe.printStackTrace(); }catch(TimeoutException toe){ toe.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); }finally{ if(channel!=null) channel.close(); if(conn!=null) conn.close(); } } }
标签：
原文地址：http://www.cnblogs.com/lyhero11/p/5128856.html
