用书籍上的文章做网站SEO,网页设计六安模板,wordpress多媒体占用id,自己建设网站需要服务器【译】RabbitMQ教程一 主要通过Hello Word对RabbitMQ有初步认识 【译】RabbitMQ教程二 工作队列#xff0c;即一个生产者对多个消费者循环分发、消息确认、消息持久、公平分发 【译】RabbitMQ教程三 如何同一个消息同时发给多个消费者开始引入RabbitMQ消息模型中的重要概念路由… 【译】RabbitMQ教程一 主要通过Hello Word对RabbitMQ有初步认识 【译】RabbitMQ教程二 工作队列即一个生产者对多个消费者循环分发、消息确认、消息持久、公平分发 【译】RabbitMQ教程三 如何同一个消息同时发给多个消费者开始引入RabbitMQ消息模型中的重要概念路由器Exchange以及绑定等使用了fanout类型的路由器 【译】RabbitMQ教程四 如何选择性地接收消息使用了direct路由器 【译】RabbitMQ教程五 如何通过多重标准接收消息使用了topic路由器可通过灵活的路由键和绑定键的设置 进一步增强消息选择的灵活性 【译】RabbitMQ教程六 如何使用RabbitMQ实现一个简单的RPC系统回调队列callback queue和关联标识correlation id 各教程代码 官方GitHub rabbitmq-tutorials 或我整理的rabbitmq-tutorial-java RabbitMQ 一般工作流程 生产者和RabbitMQ服务器建立连接和通道声明路由器同时为消息设置路由键这样所有的消息就会以特定的路由键发给路由器具体路由器会发送到哪个或哪几个队列生产者在大部分场景中都不知道。1个路由器但不同的消息可以有不同的路由键。 消费者和RabbitMQ服务器建立连接和通道然后声明队列声明路由器然后通过设置绑定键或叫路由键为队列和路由器指定绑定关系这样消费者就可以根据绑定键的设置来接收消息。1个路由器1个队列但不同的消费者可以设置不同的绑定关系。 主要方法 声明队列创建队列可以生产者和消费者都声明也可以消费者声明生产者不声明也可以生产者声明而消费者不声明。最好是都声明。生产者未声明消费者声明这种情况如果生产者先启动会出现消息丢失的情况因为队列未创建 channel.queueDeclare(String queue, //队列的名字boolean durable, //该队列是否持久化即是否保存到磁盘中boolean exclusive,//该队列是否为该通道独占的即其他通道是否可以消费该队列boolean autoDelete,//该队列不再使用的时候是否让RabbitMQ服务器自动删除掉MapString, Object arguments)//其他参数 声明路由器创建路由器生产者、消费者都要声明路由器---如果声明了队列可以不声明路由器。 channel.exchangeDeclare(String exchange,//路由器的名字String type,//路由器的类型topic、direct、fanout、headerboolean durable,//是否持久化该路由器boolean autoDelete,//是否自动删除该路由器boolean internal,//是否是内部使用的true的话客户端不能使用该路由器MapString, Object arguments) //其他参数 绑定队列和路由器只用在消费者 channel.queueBind(String queue, //队列String exchange, //路由器String routingKey, //路由键即绑定键MapString, Object arguments) //其他绑定参数 发布消息只用在生产者 channel.basicPublish(String exchange, //路由器的名字即将消息发到哪个路由器String routingKey, //路由键即发布消息时该消息的路由键是什么BasicProperties props, //指定消息的基本属性byte[] body)//消息体也就是消息的内容是字节数组 BasicProperties props指定消息的基本属性如deliveryMode为2时表示消息持久2以外的值表示不持久化消息 //BasicProperties介绍
String corrId ;
String replyQueueName ;
Integer deliveryMode 2;
String contentType application/json;
AMQP.BasicProperties props new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).deliveryMode(deliveryMode).contentType(contentType).build(); 接收消息只用在消费者 channel.basicConsume(String queue, //队列名字即要从哪个队列中接收消息boolean autoAck, //是否自动确认默认trueConsumer callback)//消费者即谁接收消息 消费者中一般会有回调方法来消费消息 Consumer consumer new DefaultConsumer(channel) {Overridepublic void handleDelivery(String consumerTag, //该消费者的标签Envelope envelope,//字面意思为信封packaging data for the messageAMQP.BasicProperties properties, //message content header data byte[] body) //message bodythrows IOException {//获取消息示例String message new String(body, UTF-8);//接下来就可以根据消息处理一些事情}}; 路由器类型 fanout会忽视绑定键每个消费者都可以接受到所有的消息前提是每个消费者都要有各自单独的队列而不是共有同一队列。direct只有绑定键和路由键完全匹配时才可以接受到消息。topic可以设置多个关键词作为路由键在绑定键中可以使用*和#来匹配headers可以忽视它的存在 教程一 HelloWorld 看主要代码 //生产者
channel.queueDeclare(QUEUE_NAME, false, false, false, null); ----①
String message Hello World!;
channel.basicPublish(, QUEUE_NAME, null, message.getBytes(UTF-8));
//消费者
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicConsume(QUEUE_NAME, true, consumer); 这里生产者和消费者都没有声明路由器而是声明了同名的队列。生产者发布消息时使用了默认的无名路由器并以队列的名字作为了路由键。消费者在消费时由于没有声明路由器这并不表示没有路由器的存在消费者此时使用的是默认的路由器即Default exchange该路由器和所有的队列都进行绑定并且使用队列的名字作为了路由键进行绑定。所以生产者使用默认路由器以队列的名字作为了绑定键进行了消息发布而消费者也使用了默认的路由器并以队列的名字作为绑定键进行了绑定。而默认路由器是direct类型路由键和绑定键完全匹配时消费者才能接受到消息所以教程1中的消费者可以接收到消息。为了认证这一点可以将代码①去掉然后先运行消费者让它等待监听然后启动生产者发送消息消费者同样会收到消息。这里的生产者声明队列只是让RabbitMQ服务器先创建这个队列以免发送的消息因为找不到队列而丢失。 教程二 Work Queues 看主要代码 //生产者
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
String message 1.;
channel.basicPublish(, TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(UTF-8));
//消费者
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
channel.basicQos(1);---①...channel.basicAck(envelope.getDeliveryTag(), false);...---③
boolean autoAck false;---②
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer); 这里也使用了默认的direct路由器。假如启动多个工作者消费者按道理这些工作者应该可以接收到所有的消息啊但是不要忘了这几个工作者都是从同一个队列中取消息消息取出一个队列中就少一个所以每个工作者都只是收到的消息的一部分。既然这几个工作者都从同一个队列中取消息那每个工作者应该怎么取呢 如果没有代码①并且②设置为true即自动确认收到消息RabbitMQ只要发出消息就认为消费者收到了此时RabbitMQ采取的是循环分发的策略在这几个工作者中循环轮流分发消息。每个工作者接受到的消息数量都是相同的。 如果有代码①并且②设置为false则RabbitMQ会采取公平分发策略即将消息发给空闲的工作者空闲工作者将消息处理完毕执行了代码③不空闲即工作者还在处理消息还没有给RabbitMQ发回确认信息即还没有执行代码③。 代码①中的参数1(prefetchCount)maximum number of messages that the server will deliver。 为了防止队列丢失在声明队列的时候指定了durable为true。为了防止消息丢失设置了消息属性BasicProperties为MessageProperties.PERSISTENT_TEXT_PLAIN让我们看看值是什么 MessageProperties.PERSISTENT_TEXT_PLAIN 可以看出里面包含了deliveryMode2。从这张图也可以看到BasicProperties属性的全貌。 如果想让多个消费者共同消费某些消息只要让他们共用同一队列即可当然前提是你得保证消息可以都进到这个队列中来如本例中使用direct路由器消息的路由键和队列的绑定键设为一致当然也可以使用fanout路由器路由键和绑定键随意设置不一致也能收到因为fanout路由器会忽略路由键的设置。 教程三 Publish/Subscribe 看主要代码 //生产者
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
channel.basicPublish(EXCHANGE_NAME, , null, message.getBytes(UTF-8));//消费者
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
String queueName channel.queueDeclare().getQueue();---①
channel.queueBind(queueName, EXCHANGE_NAME, );
channel.basicConsume(queueName, true, consumer); 教程三才引出路由器的概念。生产者和消费者声明了同样的路由并指明路由类型为fanout该路由器会忽视路由键将消息发布到所有绑定的队列中仍需要绑定只是绑定时绑定键任意就行了。 假如启动多个消费者因为代码①中调用无参的声明去恶劣方法channel.queueDeclare()就会创建了一个非持久、独特的、自动删除的队列并返回一个自动生成的名字。所以多个消费者取消息时使用的是各自的队列不会存在多个消费者从同一个队列取消息的情况。 这样多个消费者就可以接收到同一消息。 如果想实现多个消费者都可以接收到所有的消息只要让他们各自使用单独的队列即可当然前提是保证路由键和绑定键的设置可以让消息都进入到队列如本例中使用fanout路由器无需考虑绑定键和路由键。 教程4 Routing 看主要代码 //生产者
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes(UTF-8));
//消费者
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
String queueName channel.queueDeclare().getQueue();
String[] severities {info, warning, error};
for (String severity : severities) {channel.queueBind(queueName, EXCHANGE_NAME, severity);
}
channel.basicConsume(queueName, true, consumer); 可以看出教程3使用了direct路由器该路由器的特点是可以设定路由键和绑定键消费者只能从队列中取出两者匹配的消息。 在生产者发消息时为消息设置不同的路由键如例子中severity可以设为info、warn、error。 消费者在通过为队列设置多个绑定关系来选择想要接收的消息。 这里有一个概念叫做多重绑定即多个队列以相同的绑定键binding key绑定到同一个路由器上此时direct路由器就会像fanout路由器一样将消息广播给所有符合路由规则的队列。 教程5 Topics 看主要代码 //生产者
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
String routingKey ;
String message msg...;
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes(UTF-8));
//消费者
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
String queueName channel.queueDeclare().getQueue();
String bingingKeys[] {};for (String bindingKey : bingingKeys) {channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);}
channel.basicConsume(queueName, true, consumer); 这里使用了topic路由器它与direct路由器类似不同在于topic路由器可以为路由键设置多重标准。一个消息有一个路由键direct路由器只能为路由键指定一个关键字但是topic路由器可以在路由键中通过点号分割多个单词来组成路由键消费者在绑定的时候可以设置多重标准来选择接受。 举个例子假如日志根据严重级别info、warn、error也可以根据来源分为cron、kern、auth。某个日志消息设置路由键为kern.info表示来自kern的info级别的日志。想要选择接收消息的时候direct路由器就办不到它要么可以根据严重级别来筛选要么根据来源来筛选而topic路由器则可以轻松应对只要将绑定键设置为kern.info就可以精准获取该类型的日志。
