用rabbitMQ实现生产者消费者

234rfe

利用RabbitMQ实现生产者和消费者的一个小Demo
不做讲解 直接上代码

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; /** * 单消息队列通道 * Created by wangtf on 2015/11/16. * 生产者 */ public class Producer {     private  final static  String QUEUE_NAME="hello";     public static void main(String[] argv) throws Exception {         ConnectionFactory factory = new ConnectionFactory();         factory.setHost("localhost");         Connection connection = factory.newConnection();         Channel channel = connection.createChannel();         channel.queueDeclare(QUEUE_NAME, false, false, false, null);         String message = "hello world";         channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));         System.out.println(" [x] Sent '" + message + "'");         channel.close();         connection.close();     } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32

import com.rabbitmq.client.*; import java.io.IOException; /** * Created by wangtf on 2015/11/16. * 消费者 */ public class Consume {     private  final static  String QUEUE_NAME="hello";     public static void main(String[] argv) throws Exception {         ConnectionFactory factory = new ConnectionFactory();         factory.setHost("localhost");         Connection connection = factory.newConnection();         Channel channel = connection.createChannel();         channel.queueDeclare(QUEUE_NAME, false, false, false, null);         System.out.println(" Waiting for messages. To exit press CTRL+C");         Consumer consumer = new DefaultConsumer(channel) {             @Override             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)                     throws IOException {                 String message = new String(body, "UTF-8");                 System.out.println(" [x] Received '" + message + "'");             }         };         channel.basicConsume(QUEUE_NAME, true, consumer);     } }