import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Send {private final static String QUEUE_NAME = "hello";public static void main(String[] args) throws.IOException{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());System.out.println(" [x] Sent '" + message + "'");channel.close();connection.close();}
}

public class RabbitMQRecv {
　　public static void main(String avg[]) throws.IOException,java.lang.InterruptedException {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");QueueingConsumer consumer = new QueueingConsumer(channel);channel.basicConsume(QUEUE_NAME, true, consumer);while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());System.out.println(" [x] Received '" + message + "'");}}
}

什么是MQ？

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取队列中的消息。

RabbitMQ是MQ的一种。下面详细介绍一下RabbitMQ的基本概念。

1、队列、生产者、消费者

队列是RabbitMQ的内部对象，用于存储消息。生产者（下图中的P）生产消息并投递到队列中，消费者（下图中的C）可以从队列中获取消息并消费。

多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

2、Exchange、Binding

刚才我们看到生产者将消息投递到队列中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），再通过Binding将Exchange与Queue关联起来。

3、Exchange Type、Bingding key、routing key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key。在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。

RabbitMQ常用的Exchange Type有三种：fanout、direct、topic。

  fanout:把所有发送到该Exchange的消息投递到所有与它绑定的队列中。

direct:把消息投递到那些binding key与routing key完全匹配的队列中。

      topic:将消息路由到binding key与routing key模式匹配的队列中。

附上一张RabbitMQ的结构图：

最后来具体解析一下几个问题：

1、可以自动创建队列，也可以手动创建队列，如果自动创建队列，那么是谁负责创建队列呢？是生产者？还是消费者？

如果队列不存在，当然消费者不会收到任何的消息。但是如果队列不存在，那么生产者发送的消息就会丢失。所以，为了数据不丢失，消费者和生产者都可以创建队列。那么如果创建一个已经存在的队列呢？那么不会有任何的影响。需要注意的是没有任何的影响，也就是说第二次创建如果参数和第一次不一样，那么该操作虽然成功，但是队列属性并不会改变。

队列对于负载均衡的处理是完美的。对于多个消费者来说，RabbitMQ使用轮询的方式均衡的发送给不同的消费者。

2、RabbitMQ的消息确认机制

默认情况下，如果消息已经被某个消费者正确的接收到了，那么该消息就会被从队列中移除。当然也可以让同一个消息发送到很多的消费者。

如果一个队列没有消费者，那么，如果这个队列有数据到达，那么这个数据会被缓存，不会被丢弃。当有消费者时，这个数据会被立即发送到这个消费者，这个数据被消费者正确收到时，这个数据就被从队列中删除。

那么什么是正确收到呢？通过ack。每个消息都要被acknowledged（确认，ack）。我们可以显示的在程序中去ack，也可以自动的ack。如果有数据没有被ack，那么：

RabbitMQ Server会把这个信息发送到下一个消费者。

如果这个app有bug，忘记了ack，那么RabbitMQServer不会再发送数据给它，因为Server认为这个消费者处理能力有限。

而且ack的机制可以起到限流的作用（Benefitto throttling）：在消费者处理完成数据后发送ack，甚至在额外的延时后发送ack，将有效的均衡消费者的负载。

 二：代码示例

2.1：首先引入rabbitMQ jar包

 <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>3.6.5</version></dependency>

2.2：创建消费者Producer

/*** 消息生成者*/
public class Producer {public final static String QUEUE_NAME="rabbitMQ.test";public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();//设置RabbitMQ相关信息factory.setHost("localhost");//factory.setUsername("lp");//factory.setPassword("");// factory.setPort(2088);//创建一个新的连接Connection connection = factory.newConnection();//创建一个通道Channel channel = connection.createChannel();//  声明一个队列        channel.queueDeclare(QUEUE_NAME, false, false, false, null);String message = "Hello RabbitMQ";//发送消息到队列中channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println("Producer Send +'" + message + "'");//关闭通道和连接
        channel.close();connection.close();}
}

注1：queueDeclare第一个参数表示队列名称、第二个参数为是否持久化（true表示是，队列将在服务器重启时生存）、第三个参数为是否是独占队列（创建者可以使用的私有队列，断开后自动删除）、第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数

注2：basicPublish第一个参数为交换机名称、第二个参数为队列映射的路由key、第三个参数为消息的其他属性、第四个参数为发送信息的主体

2.3：创建消费者

public class Customer {private final static String QUEUE_NAME = "rabbitMQ.test";public static void main(String[] args) throws IOException, TimeoutException {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();//设置RabbitMQ地址factory.setHost("localhost");//创建一个新的连接Connection connection = factory.newConnection();//创建一个通道Channel channel = connection.createChannel();//声明要关注的队列channel.queueDeclare(QUEUE_NAME, false, false, true, null);System.out.println("Customer Waiting Received messages");//DefaultConsumer类实现了Consumer接口，通过传入一个频道，// 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDeliveryConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body)throws IOException {String message = new String(body, "UTF-8");System.out.println("Customer Received '" + message + "'");}};//自动回复队列应答 -- RabbitMQ中的消息确认机制channel.basicConsume(QUEUE_NAME, true, consumer);}

前面代码我们可以看出和生成者一样的，后面的是获取生产者发送的信息，其中envelope主要存放生产者相关信息（比如交换机、路由key等）body是消息实体。

2.4：运行结果

生产者：

消费者：

 三：实现任务分发

工作队列

一个队列的优点就是很容易处理并行化的工作能力，但是如果我们积累了大量的工作，我们就需要更多的工作者来处理，这里就要采用分布机制了。

我们新创建一个生产者NewTask

public class NewTask {private static final String TASK_QUEUE_NAME="task_queue";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory=new ConnectionFactory();factory.setHost("localhost");Connection connection=factory.newConnection();Channel channel=connection.createChannel();channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);//分发信息for (int i=0;i<10;i++){String message="Hello RabbitMQ"+i;channel.basicPublish("",TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());System.out.println("NewTask send '"+message+"'");}channel.close();connection.close();}
}

然后创建2个工作者Work1和Work2代码一样

public class Work1 {private static final String TASK_QUEUE_NAME = "task_queue";public static void main(String[] args) throws IOException, TimeoutException {final ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();final Channel channel = connection.createChannel();channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println("Worker1  Waiting for messages");//每次从队列获取的数量channel.basicQos(1);final Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("Worker1  Received '" + message + "'");try {throw  new Exception();//doWork(message);}catch (Exception e){channel.abort();}finally {System.out.println("Worker1 Done");channel.basicAck(envelope.getDeliveryTag(),false);}}};boolean autoAck=false;//消息消费完成确认channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);}private static void doWork(String task) {try {Thread.sleep(1000); // 暂停1秒钟} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}
}

注：channel.basicQos(1);保证一次只分发一个 。autoAck是否自动回复，如果为true的话，每次生产者只要发送信息就会从内存中删除，那么如果消费者程序异常退出，那么就无法获取数据，我们当然是不希望出现这样的情况，所以才去手动回复，每当消费者收到并处理信息然后在通知生成者。最后从队列中删除这条信息。如果消费者异常退出，如果还有其他消费者，那么就会把队列中的消息发送给其他消费者，如果没有，等消费者启动时候再次发送。

参考：https://www.cnblogs.com/LipeiNet/p/5977028.html