Java RabbitMQ 完整示例 带联接断开恢复,发布接收完整示例
import java.io.IOException; import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; 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; import com.rabbitmq.client.QueueingConsumer.Delivery; import com.rabbitmq.client.ShutdownSignalException; /* * @rabbitmq数据测试 * */ public class RabbitMQTest { // 队列名称 private final static String QUEUE_NAME = "hello"; // 创建对像 ConnectionFactory factory; // 联接信息 Connection connection; // 频道 Channel channel = null; // 获取数据队列 QueueingConsumer consumer = null; /** * InitMQ 初始化MQ信息 * * * */ public void InitMQ() { // 打开连接和创建频道，与发送端一样 factory = new ConnectionFactory(); factory.setHost("192.168.2.110"); factory.setUsername("rabbitmq"); factory.setPassword("rabbitmq"); factory.setVirtualHost("/"); factory.setPort(5672); // 关键所在，指定线程池 ExecutorService service = Executors.newFixedThreadPool(10); factory.setSharedExecutor(service); // 设置自动恢复 factory.setAutomaticRecoveryEnabled(true); factory.setNetworkRecoveryInterval(2);// 设置 没10s ，重试一次 factory.setTopologyRecoveryEnabled(false);// 设置不重新声明交换器，队列等信息。 try { connection = factory.newConnection(); channel = connection.createChannel(); // 声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out .println(" [*] Waiting for messages. To exit press CTRL+C"); // 设置一条条的应答 channel.basicQos(0, 1, false); // 创建队列消费者 consumer = new QueueingConsumer(channel); // 指定消费队列 ,第二个参数false表示不自动应答 channel.basicConsume(QUEUE_NAME, false, consumer); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (TimeoutException e) { // TODO Auto-generated catch block e.printStackTrace(); } } /** * CycleSend 调用线程池循环发送数据<br/> * * * */ public void CycleSend() { // 使用线程池 ScheduledExecutorService scheduledThreadPool = Executors .newScheduledThreadPool(5); // 定时执行任务 scheduledThreadPool.scheduleAtFixedRate(new Runnable() { @Override public void run() { // 发送的消息 String message = "hello world!" + UUID.randomUUID(); // TODO Auto-generated method stub try { if (!channel.isOpen()) { // 初始化恢复 // InitMQ(); // System.out.println(" [x] 启动发送重联"+channel.isOpen()); } if (channel.isOpen()) { channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); } } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }, 1, 3, TimeUnit.SECONDS); } /** * ReceiveMQ 从MQ中收取数据 <br/> * 并且可以收获MQ信息数据 * * */ public void ReceiveMQ() { Thread thr = new Thread(new Runnable() { @Override public void run() { Cycle(); } }); thr.start(); } // 是否断开 boolean isBroken = false; // 循环读取数据 protected void Cycle() { while (true) { // nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法） QueueingConsumer.Delivery delivery; try { if (isBroken) { //channel.basicRecover(); System.out.println(" [x] 开始接收数据可能会卡 ----"+ channel.isOpen()); } if (channel.isOpen()) { delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); //应答 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); System.out.println(" [x] Received '" + message + "'"); } } catch (ShutdownSignalException e) { try { if (channel.isOpen()) { // 异常情况恢复 consumer = new QueueingConsumer(channel); channel.basicConsume(QUEUE_NAME, false, consumer); isBroken = false; System.out.println(" [x] 恢复成功 ----"); } } catch (Exception e1) { isBroken = true; // TODO Auto-generated catch block e1.printStackTrace(); System.out.println(" [x] 接收线程--- ShutdownSignalException-Exception"); } // TODO Auto-generated catch block // e.printStackTrace(); } catch (ConsumerCancelledException e) { // TODO Auto-generated catch block e.printStackTrace(); System.out.println(" [x] 接收线程跳----ConsumerCancelledException"); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); System.out.println(" [x] 接收线程跳出 ----Exception"); break; } } } }
