rabbitMQ-demo.zip_DEMO_piguhw_rabbitMQ-demo_rabbitmq .idx

package com.xsl.rabbitmq.workQueues; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; import java.util.Map; /*多个消费者对同一个队列进行消费*/ public class WorkQueueConsumer { private static final String QUEUE_NAME = "hello"; public static void main(String[] args) { new WorkQueueConsumer(). new ReceiveThread().start(); new WorkQueueConsumer(). new ReceiveThread().start(); } class ReceiveThread extends Thread { @Override public void run() { try { ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672); connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); boolean durable = true; boolean exclusive = false; boolean autoDelete = false; Map arguments = null; channel.queueDeclare(QUEUE_NAME,durable,exclusive,autoDelete,arguments); QueueingConsumer consumer = new QueueingConsumer(channel); boolean autoAck = false; boolean multi = false; channel.basicConsume(QUEUE_NAME,autoAck,consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.printf(Thread.currentThread().getName()+" Receive message :"+message+"\n"); Thread.sleep(3000); channel.basicAck(delivery.getEnvelope().getDeliveryTag(),multi); } } catch (Exception e) { } } } }