基本概念
ConnectionFactory、Connection、Channel
ConnectionFactory、Connection、Channel，这三个都是RabbitMQ对外提供的API中最基本的对象。不管是服务器端还是客户端都会首先创建这三类对象。
ConnectionFactory为Connection的制造工厂。Connection是与RabbitMQ服务器的socket链接，它封装了socket协议及身份验证相关部分逻辑。
Channel是我们与RabbitMQ打交道的最重要的一个接口，大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。
Queue
Queue（队列）是RabbitMQ的内部对象，用于存储消息
RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者可以从Queue中获取消息并消费。
队列是有Channel声明的，而且这个操作是幂等的。同名的队列多次声明也只会创建一次。我们发送消息就是想这个声明的队列里发送消息
工作队列的主要思想是不用等待资源密集型的任务处理完成，
为了确保消息或者任务不会丢失，rabbitmq 支持消息确信 ACK。ACK机制是消费者端从rabbitmq收到消息并处理完成后，反馈给rabbitmq，rabbitmq收到反馈信息后将消息从队列中删除
如果rabbitmq向消费者改善消息时，消费者服务器挂了，消息也不会超时，即使一个消息需要非常长的时间处理，也不会导致消息超时，永远不会从rabbitmq中删除，
忘记通过basicAck返回确认信息是个严重的错误
rabbitmq不允许重新定义一个已有的队列信息
QueueingConsumer
队列消费者，用于监听队列中的消息。调用nextDelivery方法时，内部实现就是调用队列的take方法。该方法的作用：获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）。说白了就是如果没有消息，就处于阻塞状态。
RabbitMQ 笔记-Exchanges
Procuder Publish的Message进入了Exchange。接着通过“routing keys”， RabbitMQ会找到应该把这个Message放到哪个queue里。queue也是通过这个routing keys来做的绑定。
有三种类型的Exchanges：direct, fanout,topic。 每个实现了不同的路由算法（routing algorithm）。
Direct exchange: 如果 routing key 匹配, 那么Message就会被传递到相应的queue中。其实在queue创建时，它会自动的以queue的名字作为routing key来绑定那个exchange。
生产者： // 声明交换器 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 发送消息 for (String severity : routingKeys) { String message = "Send the message level: " + severity; channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes()); System.out.println(" [x] Sent '" + severity + "':'" + message + "'"); }消费者 // 声明交换器 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 获取匿名队列名称 String queueName = channel.queueDeclare().getQueue(); // 根据路由关键字进行多重绑定 for (String severity : routingKeys) { channel.queueBind(queueName, EXCHANGE_NAME, severity); System.out.println("ReceiveLogsDirect1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity); }
Fanout exchange: 会向响应的queue广播。
生产者 channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 分发消息 for (int i=0; i<5; i++) { String message = "Hello World!" + i; channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); }消费者 channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, "");
Topic exchange: 对key进行模式匹配，比如ab*可以传递到所有ab*的queue。
生产者 // 声明一个匹配模式的交换器 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 待发送的消息 String routingKeys[] = new String[]{ "quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox", "lazy.brown.fox", "quick.brown.fox", "quick.orange.male.rabbit", "lazy.orange.male.rabbit" }; // 发送消息 for (String severity : routingKeys) { String message = "From "+severity+" routingKey' s message!"; channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes()); System.out.println("TopicSend [x] Sent '" + severity + "':'" + message + "'"); }消费者 // 声明一个匹配模式的交换器 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); String queueName = channel.queueDeclare().getQueue(); // 路由关键字 String routingKeys[] = new String[] { "*.orange.*" }; // 绑定路由关键字 for (String bindingKey : routingKeys) { channel.queueBind(queueName, EXCHANGE_NAME, bindingKey); System.out.println("ReceiveLogsTopic1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + bindingKey); } 匿名: 直接发送到queue。
生产者 for (int i=0; i<5; i++) { String message = "hello world! " + i; channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); }消费者 channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); RabbitMQ 笔记-RPC
RabbitMQ中实现RPC的机制是：
- 客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
- 服务器端收到消息并处理
- 服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
- 客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理
public class RPCClient { private static final String RPC_QUEUE_NAME = "rpc_queue"; private Connection connection; private Channel channel; private String replyQueueName; private QueueingConsumer consumer; public RPCClient() throws Exception { ConnectionFactory factory = new ConnectionFactory(); // 设置MabbitMQ所在主机ip或者主机名 factory.setHost("192.168.65.136"); factory.setUsername("rabbitmq"); factory.setPassword("123456"); // 创建一个连接 connection = factory.newConnection(); // 创建一个频道 channel = connection.createChannel(); //声明队列 channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); //为每一个客户端获取一个随机的回调队列 replyQueueName = channel.queueDeclare().getQueue(); //为每一个客户端创建一个消费者（用于监听回调队列，获取结果） consumer = new QueueingConsumer(channel); //消费者与队列关联 channel.basicConsume(replyQueueName, true, consumer); } /** * 获取斐波列其数列的值 * * @param message * @return * @throws Exception */ public String call(String message) throws Exception{ String response = null; String corrId = java.util.UUID.randomUUID().toString(); //设置replyTo和correlationId属性值 BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build(); //发送消息到rpc_queue队列 channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes()); while (true) { System.out.println("OK?"); QueueingConsumer.Delivery delivery = consumer.nextDelivery(); System.out.println("OK"); if (delivery.getProperties().getCorrelationId().equals(corrId)) { response = new String(delivery.getBody(),"UTF-8"); break; } } return response; } public static void main(String[] args) throws Exception { RPCClient fibonacciRpc = new RPCClient(); String result = fibonacciRpc.call("4"); System.out.println( "fib(4) is " + result); }} public class RPCServer { private static final String RPC_QUEUE_NAME = "rpc_queue"; public static void main(String[] args) throws Exception { ConnectionFactory factory = new ConnectionFactory(); // 设置MabbitMQ所在主机ip或者主机名 factory.setHost("192.168.65.136"); factory.setUsername("rabbitmq"); factory.setPassword("123456"); // 创建一个连接 Connection connection = factory.newConnection(); // 创建一个频道 Channel channel = connection.createChannel(); //声明队列 channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); //限制：每次最多给一个消费者发送1条消息 channel.basicQos(1); //为rpc_queue队列创建消费者，用于处理请求 QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(RPC_QUEUE_NAME, false, consumer); System.out.println(" [x] Awaiting RPC requests"); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); //获取请求中的correlationId属性值，并将其设置到结果消息的correlationId属性中 BasicProperties props = delivery.getProperties(); BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build(); //获取回调队列名字 String callQueueName = props.getReplyTo(); String message = new String(delivery.getBody(),"UTF-8"); System.out.println(" [.] fib(" + message + ")"); //获取结果 String response = "" + fib(Integer.parseInt(message)); //先发送回调结果 channel.basicPublish("", callQueueName, replyProps,response.getBytes()); //后手动发送消息反馈 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } /** * 计算斐波列其数列的第n项 * * @param n * @return * @throws Exception */ private static int fib(int n) throws Exception { if (n < 0) throw new Exception("参数错误，n必须大于等于0"); if (n == 0) return 0; if (n == 1) return 1; return fib(n - 1) + fib(n - 2); }}