rabbitmq

基础：

工作队列主要是为了避免资源密集型任务的立即执行，然后一直等待它执行结束。相反，我们可以安排好任务，然后在执行。我们可以将一个任务封装成一个消息，发送到队列中。由工作者在后台取出任务然后执行。当有多个工作者时，他们共同处理这些任务。

 

demo:

package rabbitmq;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

/**
 * 消费者从消息队列取消息
 * @author thrillerzw
 *
 */
public class Reqv {
 private final static String QUEUE_NAME = "hello";

 public static void main(String[] argv) throws InterruptedException, IOException {

  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("192.168.0.200");
  // factory.setPort(15672);
  factory.setUsername("thrillerzw");
  factory.setPassword("123456");
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
  //parameters for queue 'hello4' in vhost '/' not equivalent
  boolean durable=true;
  channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
  System.out.println("Waiting for messages.");
  QueueingConsumer consumer = new QueueingConsumer(channel);
  //事务方式： boolean autoAck ：false,需要手工应答，mq才会认为消费成功。
  boolean autoAck=false;
  String consumerTag = channel.basicConsume(QUEUE_NAME, autoAck, consumer);
  System.out.println("consumerTag=" + consumerTag);
        int i=0;
  while (true) {
   QueueingConsumer.Delivery delivery = consumer.nextDelivery();
   String message = new String(delivery.getBody());
   try {
   //业务
   System.out.println("Received '" + message + "'");
   Thread.sleep(100);
   if(i==2){
     throw new Exception(); 
   }
   //autoAck ：false时候应答确认消息处理完成
   long deliveryTag=delivery.getEnvelope().getDeliveryTag();
   channel.basicAck(deliveryTag, false);
   System.out.println();
   } catch (Throwable e) {
    e.printStackTrace();
    //测试看是回滚到队列的末尾.
    //不断回滚可能出现死循环？
    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
   }
   i++;
  }
 }
}

package rabbitmq;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

/**
 * 生产者向消息队列发送消息  
 * 防止宕机重启内存丢消息， 持久化队列。
 * channel.queueDeclare设置 boolean durable：true,channel.basicPublish消息也必须标记为MessageProperties.PERSISTENT_TEXT_PLAIN 
 * @author thrillerzw
 *
 */
public class Send {
 private final static String QUEUE_NAME = "hello";

 public static void main(String[] args) throws IOException {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("192.168.0.200");
  factory.setUsername("thrillerzw");
  factory.setPassword("123456");

  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
  //程序声明一个队列,如果没有会添加。持久化队列：durable：true
  channel.queueDeclare(QUEUE_NAME, true, false, false, null);
  for(int i=0;i<10;i++){
   String message = "Hello World!"+i;
   //持久化消息：MessageProperties.PERSISTENT_TEXT_PLAIN 非持久化： MessageProperties.TEXT_PLAIN
   channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,
     message.getBytes());
   System.out.println("Sent '" + message + "'");
  }
  channel.close();
  connection.close();
 }
}

package rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * @author thrillerzw
 *
 */
public class SendLogs {
 private static final String EXCHANGE_NAME = "logs";

 public static void main(String[] argv) throws Exception {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("192.168.0.200");
  factory.setUsername("thrillerzw");
  factory.setPassword("123456");
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
  channel.exchangeDeclare(EXCHANGE_NAME, "fanout");//声明Exchange
  for (int i = 0; i <= 2; i++) {
   String message = "hello word!" + i;
   channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
   System.out.println("Sent '" + message + "'");
  }
  channel.close();
  connection.close();
 }

}

package rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class ReceiveLogs {
 private static final String EXCHANGE_NAME = "logs";

 public static void main(String[] argv) throws Exception {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("192.168.0.200");
  factory.setUsername("thrillerzw");
  factory.setPassword("123456");
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
        
  channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
/*	 String queueName = "log-fb1";
  //通过channel.queueDeclare()来创建一个非持久的队列名
  channel.queueDeclare(queueName, false, false, false, null);*/
  //rabbitmq创建一个随机的auto-delete的队列，并返回名字。amq.gen-iieJpst2mQghdrXF7nKYCQ / amq.gen-iieJpst2mQghdrXF7nKYCQ
  String queueName = channel.queueDeclare().getQueue();
  channel.queueBind(queueName, EXCHANGE_NAME, "");//把Queue、Exchange绑定
  QueueingConsumer consumer = new QueueingConsumer(channel);
  channel.basicConsume(queueName, true, consumer);
  while (true) {
   QueueingConsumer.Delivery delivery = consumer.nextDelivery();
   String message = new String(delivery.getBody());
   System.out.println("Received '" + message + "'");
  }
 }
}