作者：海向
来源：
rabbitmq 简述
rabbitmq是一个消息代理：它接受并转发消息。 您可以将其视为邮局：当您将要把寄发的邮件投递到邮箱中时，您可以确信postman 先生最终会将邮件发送给收件人。 在这个比喻中，rabbitmq是一个邮箱，邮局和邮递员，用来接受，存储和转发二进制数据块的消息。
队列就像是在rabbitmq中扮演邮箱的角色。 虽然消息经过rabbitmq和应用程序，但它们只能存储在队列中。 队列只受主机的内存和磁盘限制的限制，它本质上是一个大的消息缓冲区。 许多生产者可以发送到一个队列的消息，许多消费者可以尝试从一个队列接收数据。
producer即为生产者，用来产生消息发送给队列。consumer是消费者，需要去读队列内的消息。producer，consumer和broker(rabbitmq server)不必驻留在同一个主机上;确实在大多数应用程序中它们是这样分布的。
简单队列
简单队列是最简单的一种模式，由生产者、队列、消费者组成。生产者将消息发送给队列，消费者从队列中读取消息完成消费。
在下图中，“p”是我们的生产者，“c”是我们的消费者。 中间的框是队列 – rabbitmq代表消费者的消息缓冲区。
java 方式
生产者
package com.anqi.mq.nat;
import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
public class myproducer {
private static final string queue_name = "item_queue";
public static void main(string[] args) throws exception {
//1. 创建一个 connectionfactory 并进行设置
connectionfactory factory = new connectionfactory();
factory.sethost("localhost");
factory.setvirtualhost("/");
factory.setusername("guest");
factory.setpassword("guest");
//2. 通过连接工厂来创建连接
connection connection = factory.newconnection();
//3. 通过 connection 来创建 channel
channel channel = connection.createchannel();
//实际场景中，消息多为json格式的对象
string msg = "hello";
//4. 发送三条数据
for (int i = 1; i <= 3 ; i++) {
channel.basicpublish("", queue_name, null, msg.getbytes());
system.out.println("send message" + i +" : " + msg);
}
//5. 关闭连接
channel.close();
connection.close();
}
}
/**
* declare a queue
* @param queue the name of the queue
* @param durable true if we are declaring a durable queue (the queue will survive a server restart)
* @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
* @param autodelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
* @param arguments other properties (construction arguments) for the queue
* @return a declaration-confirm method to indicate the queue was successfully declared
* @throws java.io.ioexception if an error is encountered
*/
queue.declareok queuedeclare(string queue, boolean durable, boolean exclusive, boolean autodelete,map<string, object> arguments) throws ioexception;
/**
* publish a message
* @see com.rabbitmq.client.amqp.basic.publish
* @param exchange the exchange to publish the message to
* @param routingkey the routing key
* @param props other properties for the message - routing headers etc
* @param body the message body
* @throws java.io.ioexception if an error is encountered
*/
void basicpublish(string exchange, string routingkey, basicproperties props, byte[] body) throws ioexception;
/**
* start a non-nolocal, non-exclusive consumer, with
* a server-generated consumertag.
* @param queue the name of the queue
* @param autoack true if the server should consider messages
* acknowledged once delivered; false if the server should expect
* explicit acknowledgements
* @param callback an interface to the consumer object
* @return the consumertag generated by the server
* @throws java.io.ioexception if an error is encountered
* @see com.rabbitmq.client.amqp.basic.consume
* @see com.rabbitmq.client.amqp.basic.consumeok
* @see #basicconsume(string, boolean, string, boolean, boolean, map, consumer)
*/
string basicconsume(string queue, boolean autoack, consumer callback) throws ioexception;
消费者
package com.anqi.mq.nat;
import com.rabbitmq.client.*;
import java.io.ioexception;
public class myconsumer {
private static final string queue_name = "item_queue";
public static void main(string[] args) throws exception {
//1. 创建一个 connectionfactory 并进行设置
connectionfactory factory = new connectionfactory();
factory.sethost("localhost");
factory.setvirtualhost("/");
factory.setusername("guest");
factory.setpassword("guest");
//2. 通过连接工厂来创建连接
connection connection = factory.newconnection();
//3. 通过 connection 来创建 channel
channel channel = connection.createchannel();
//4. 声明一个队列
channel.queuedeclare(queue_name, true, false, false, null);
system.out.println(" [*] waiting for messages. to exit press ctrl+c");
/*
true:表示自动确认，只要消息从队列中获取，无论消费者获取到消息后是否成功消费，都会认为消息已经成功消费
false:表示手动确认，消费者获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一
直没有反馈，那么该消息将一直处于不可用状态，并且服务器会认为该消费者已经挂掉，不会再给其发送消息，
直到该消费者反馈。
*/
//5. 创建消费者并接收消息
consumer consumer = new defaultconsumer(channel) {
@override
public void handledelivery(string consumertag, envelope envelope,
amqp.basicproperties properties, byte[] body)
throws ioexception {
string message = new string(body, "utf-8");
system.out.println(" [x] received '" + message + "'");
}
};
//6. 设置 channel 消费者绑定队列
channel.basicconsume(queue_name, true, consumer);
}
}
send message1 : hello
send message2 : hello
send message3 : hello
[*] waiting for messages. to exit press ctrl+c
[x] received 'hello'
[x] received 'hello'
[x] received 'hello'
当我们启动生产者之后查看rabbitmq管理后台可以看到有一条消息正在等待被消费。
当我们启动消费者之后再次查看，可以看到积压的一条消息已经被消费。
总结
队列声明queuedeclare的参数：第一个参数表示队列名称、第二个参数为是否持久化（true表示是，队列将在服务器重启时生存）、第三个参数为是否是独占队列（创建者可以使用的私有队列，断开后自动删除）、第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数。
basicconsume的第二个参数autoack: 应答模式，true：自动应答，即消费者获取到消息，该消息就会从队列中删除掉，false：手动应答，当从队列中取出消息后，需要程序员手动调用方法应答，如果没有应答，该消息还会再放进队列中，就会出现该消息一直没有被消费掉的现象。
这种简单队列的模式，系统会为每个队列隐式地绑定一个默认交换机，交换机名称为” (amqp default)”，类型为直连 direct，当你手动创建一个队列时，系统会自动将这个队列绑定到一个名称为空的 direct 类型的交换机上，绑定的路由键 routing key 与队列名称相同，相当于channel.queuebind(queue:"queue_name", exchange:"(amqp default)“, routingkey:"queue_name");虽然实例没有显式声明交换机，但是当路由键和队列名称一样时，就会将消息发送到这个默认的交换机中。这种方式比较简单，但是无法满足复杂的业务需求，所以通常在生产环境中很少使用这种方式。
the default exchange is implicitly bound to every queue, with a routing key equal to the queue name. it is not possible to explicitly bind to, or unbind from the default exchange. it also cannot be deleted.默认交换机隐式绑定到每个队列，其中路由键等于队列名称。不可能显式绑定到，或从缺省交换中解除绑定。它也不能被删除。
——引自 rabbitmq 官方文档
spring-amqp方式
引入 maven 依赖
<dependency>
<groupid>com.rabbitmq</groupid>
<artifactid>amqp-client</artifactid>
<version>5.6.0</version>
</dependency>
<dependency>
<groupid>org.springframework.amqp</groupid>
<artifactid>spring-rabbit</artifactid>
<version>2.1.5.release</version>
</dependency>
spring 配置文件
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemalocation="http://www.springframework.org/schema/rabbit
https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd">
<rabbit:connection-factory id="connectionfactory" host="localhost" virtual-host="/"
username="guest" password="guest"/>
<rabbit:template id="amqptemplate" connection-factory="connectionfactory"/>
<rabbit:admin connection-factory="connectionfactory"/>
<rabbit:queue name="my-queue"/>
</beans>
使用测试
import org.springframework.amqp.core.amqptemplate;
import org.springframework.context.applicationcontext;
import org.springframework.context.support.classpathxmlapplicationcontext;
public class main {
public static void main(string[] args) {
applicationcontext app = new classpathxmlapplicationcontext("spring/rabbit-context.xml");
amqptemplate amqptemplate = app.getbean(amqptemplate.class);
amqptemplate.convertandsend("my-queue", "item");
string msg = (string) amqptemplate.receiveandconvert("my-queue");
system.out.println(msg);
}
}
参考方法
/**
* convert a java object to an amqp {@link message} and send it to a specific exchange
* with a specific routing key.
*
* @param exchange the name of the exchange
* @param routingkey the routing key
* @param message a message to send
* @throws amqpexception if there is a problem
*/
void convertandsend(string exchange, string routingkey, object message) throws amqpexception;
/**
* receive a message if there is one from a specific queue and convert it to a java
* object. returns immediately, possibly with a null value.
*
* @param queuename the name of the queue to poll
* @return a message or null if there is none waiting
* @throws amqpexception if there is a problem
*/
@nullable
object receiveandconvert(string queuename) throws amqpexception;
近期热文推荐：
1.1,000+ 道 java面试题及答案整理(2021最新版)
2.终于靠开源项目弄到 intellij idea 激活码了，真香！
3.阿里 mock 工具正式开源，干掉市面上所有 mock 工具！
4.spring cloud 2020.0.0 正式发布，全新颠覆性版本！
5.《java开发手册（嵩山版）》最新发布，速速下载！
觉得不错，别忘了随手点赞+转发哦！
