河南新乡做网站公司哪家好,新能源网站开发,企业网站定制开发流程,828企业服务平台文章目录 一、第一种模型#xff08;Hello World#xff09;二、第二种模型#xff08;work queue#xff09;自动确认机制的后果和公平分配 三、阐述默认交换机 这篇博客是以下资料学后的总结#xff1a; 不良人的RabbitMQ的教学视频 官方启动教程 RabbitMQ中文文档
一、… 文章目录 一、第一种模型Hello World二、第二种模型work queue自动确认机制的后果和公平分配 三、阐述默认交换机 这篇博客是以下资料学后的总结 不良人的RabbitMQ的教学视频 官方启动教程 RabbitMQ中文文档
一、第一种模型Hello World 在上图的模型中有以下概念
P生产者也就是要发送消息的程序C消费者消息的接受者会一直等待消息的到来。queue消息队列图中红色部分。类似于一个邮箱可以缓存消息生产者向其中投递消息消费者从其中取出消息。
开发生产者 // 创建连接mq的连接工厂对象ConnectionFactory connectionFactory new ConnectionFactory();// 设置连接Rabbitmq的主机connectionFactory.setHost(192.168.248.135);// 设置端口号connectionFactory.setPort(5672);// 设置连接那个虚拟主机connectionFactory.setVirtualHost(/ems);// 设置用户名和密码connectionFactory.setUsername(ems);connectionFactory.setPassword(123456);// 获取连接对象Connection connection connectionFactory.newConnection();// 获取连接中的通道对象Channel channel connection.createChannel();// 通过通道绑定对应的消息队列// 参数1队列的名称 如果队列不存在会自动创建// 参数2用来定义队列特性是否需要持久化true:持久化队列false即不持久化// 参数3exclusive 是否独占队列// 参数4是否在消费完成后自动删除队列// 参数5额外参数// 这个不加是没关系的只是表示我的Rabbitmq中是有hello消息队列的消费者产生的
channel.queueDeclare(hello,false,false,false,null);// 发布消息// 参数1交换机名称参数2路由键名称参数3传递消息额外设置参数4消息的具体内容channel.basicPublish(,hello,null,hello rabbitmq.getBytes());channel.close();connection.close();消费者开始消费 // 创建连接工厂ConnectionFactory connectionFactory new ConnectionFactory();// 设置主机和端口connectionFactory.setHost(192.168.248.135);connectionFactory.setPort(5672);// 设置虚拟主机connectionFactory.setVirtualHost(/ems);// 设置用户名和密码connectionFactory.setUsername(ems);connectionFactory.setPassword(123456);// 创建连接对象Connection connection connectionFactory.newConnection();// 创建通道Channel channel connection.createChannel();// 通道绑定对象channel.queueDeclare(hello,false,false,false,null);// 消费消息// 参数1消费哪个队列的消息 队列名称// 参数2开始消息的自动确认机制// 参数3消费消息时的回调接口String hello channel.basicConsume(hello, true, new DefaultConsumer(channel){// 最后一个参数消息队列中取出的消息Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {System.out.println(转化成对应的字符串 new String(body));System.out.println();}});/*channel.close();connection.close();*/这里需要注意由于这里是多线程下生产、消费消息所以在消费时不应该提前关闭通道不然无法监听到队列中的数据。
下面是证明看看各线程的名称就知道了。 需注意Junit5之前是不支持多线程的。
参数说明
channel.queueDeclare(hello,true,false,true,null);
参数1用来声明通道对应的队列
参数2用来指定是否持久化队列
参数3用来指定是否独占队列一般为false
参数4用来指定是否自动删除队列
参数5对队列的额外配置参数1交换机名称参数2路由键名称参数3传递消息额外设置参数4消息的具体内容
channel.basicPublish(,hello, MessageProperties.PERSISTENT_TEXT_PLAIN,I love you~.getBytes());二、第二种模型work queue
Work Queues也被称为Task Queues 任务模型。当消息处理比较耗时的时候可能生产消息的速度会远远大于消息的消费速度。长此以往消息就会堆积越来越多无法及时处理。此时就可以使用 work 模型让多个消费者绑定到一个队列共同消费队列中的消息。队列中的消息一旦消费就会消失因此任务是不会被重复执行的。 角色
Sender生产者任务的发布者Consumer消费者领取任务并且完成任务
生产者代码 // 获取连接对象Connection conn RabbitMQUtils.getConnection();// 获取通道对象Channel channel conn.createChannel();// 通过通道声明队列channel.queueDeclare(work, true, false, false, null);for (int i 0; i 10; i) {// 生产消息channel.basicPublish(, work, null, (( i )Hello worke queue~).getBytes());}// 关闭资源RabbitMQUtils.closeConnectionAndChanel(conn, channel);消费者代码
Runnable myRunnable new Runnable() {Overridepublic void run() {Connection conn RabbitMQUtils.getConnection();try {final int[] cnt new int[1];cnt[0] 0;Channel channel conn.createChannel();channel.queueDeclare(work, true, false, false, null);System.out.println(当前线程 Thread.currentThread().getName());channel.basicConsume(work, true, new DefaultConsumer(channel) {Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {System.out.println(消费者- consumerTag new String(body));System.out.println();cnt[0];}});System.out.println(Thread.currentThread().getName() : cnt[0]);} catch (Exception e) {e.printStackTrace();}}};Thread work1 new Thread(myRunnable, work-001);Thread work2 new Thread(myRunnable, work-002);Thread work3 new Thread(myRunnable, work-003);work1.start();work2.start();work3.start();测试结果 总结默认情况下RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言每个消费者都会收到相同数量的消息。这种分发消息的方式称为轮询。如下图所示 下面是官方给的 By default, RabbitMQ will send each message to the next consumer, in sequence. On average every consumer will get the same number of messages. This way of distributing messages is called round-robin. Try this out with three or more workers. 自动确认机制的后果和公平分配
channel.basicConsume(work, true, new DefaultConsumer(channel)
// 这里的第二个参数是指是否默认提交上面 work queue 实现会出现两个问题
问题一这里将 autoAcked 参数值设置为了 true即消费者收到消息队列调度的消息后不管有没有消费成功都立即返回 ACK 确认消息队列只顾着轮询分配去了。这个时候的话会引发一个问题当消费者突然宕机了那还没处理的消息就不会被处理即消失了。比如一个消费者被分配到了五个消息但是只处理了三个就嘎了那剩下的俩个就处理不了了。
问题二在两个worker的情况下当所有奇数消息都很重偶数消息都很轻时一个worker将一直很忙而另一个几乎不做任何工作。但是RabbitMQ对此一无所知仍然会均匀地分发消息。我们应该遵循能者多劳充分利用资源但轮询方式总是这么的不合我们的胃口。
解决方案 首先得将自动提交设置为 false手动提交就好了 每一次给空闲发消费者一个消息即设置 prefetchCount 1这样的话不会让能者出现不工作懒者一堆事没做的情况。当消费者死亡即通道关闭、连接被关闭、或者TCP连接丢失等情况还没有发送ACK那有其他消费者在线的话消息队列会将消息迅速交付给另一个消费者从而确保消息没有丢失。 具体解决方案的伪代码如下
// 配置每一次只能执行一个小希
channel.basicQos(1);
// 关闭手动提交
channel.basicConsume(work, false, new DefaultConsumer(channel) {Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {System.out.println(消费者- consumerTag new String(body));// 参数1确认队列中哪些具体消息 参数2是否开启多个消息同时确认channel.basicAck(envelope.getDeliveryTag(), false);}
});三、阐述默认交换机
可以在 RabbitMQ中文文档-默认交换机 去了解更多AMQP协议的一些内容。
默认交换机的本质是直连交换机当你添加一个队列的时候这个队列第一反应就是绑定默认交换机而绑定binding的路由键名称和队列名称是一致的。
上面两种模型RabbitMQ官方教程阐述的Hello World模型和Work Queues模型在官方教程中没有指出使用了交换机但是本质都是绑定了默认交换机的也就是直连交换机它也是支持多消费者的负载均衡的。
首先必须知道的是使用默认交换机时队列是在消费者端创建的可以说是用户本身吧而不是生产者去创建的。当生产者发送一条消息到 RabbitMQ 时RabbitMQ 会根据消息的路由键在使用默认交换机的情况下路由键即为队列名称来查找是否已经存在该队列如果队列不存在则会丢弃该消息。
从 basicPublish 方法的第二个参数为路由键名称和 basicConsume 方法的第一个参数为队列名称也是可以看出来的。
总的来说就是生产者不需要关心队列的创建这是消费者需要声明指定的默认交换机会绑定声明的消息队列的所以生产者该发发创建的任务就不用管了。 文章转载自： http://www.morning.kndt.cn.gov.cn.kndt.cn http://www.morning.yrwqz.cn.gov.cn.yrwqz.cn http://www.morning.qtltg.cn.gov.cn.qtltg.cn http://www.morning.cctgww.cn.gov.cn.cctgww.cn http://www.morning.gsjw.cn.gov.cn.gsjw.cn http://www.morning.ckbmz.cn.gov.cn.ckbmz.cn http://www.morning.mcjrf.cn.gov.cn.mcjrf.cn http://www.morning.hfxks.cn.gov.cn.hfxks.cn http://www.morning.qcslh.cn.gov.cn.qcslh.cn http://www.morning.yqrgq.cn.gov.cn.yqrgq.cn http://www.morning.nmkbl.cn.gov.cn.nmkbl.cn http://www.morning.wkcl.cn.gov.cn.wkcl.cn http://www.morning.btmwd.cn.gov.cn.btmwd.cn http://www.morning.pdwzr.cn.gov.cn.pdwzr.cn http://www.morning.cprbp.cn.gov.cn.cprbp.cn http://www.morning.kpxzq.cn.gov.cn.kpxzq.cn http://www.morning.lwqst.cn.gov.cn.lwqst.cn http://www.morning.gmgyt.cn.gov.cn.gmgyt.cn http://www.morning.nlkm.cn.gov.cn.nlkm.cn http://www.morning.ccpnz.cn.gov.cn.ccpnz.cn http://www.morning.lpmdy.cn.gov.cn.lpmdy.cn http://www.morning.njdtq.cn.gov.cn.njdtq.cn http://www.morning.rcwzf.cn.gov.cn.rcwzf.cn http://www.morning.gwtbn.cn.gov.cn.gwtbn.cn http://www.morning.mxxsq.cn.gov.cn.mxxsq.cn http://www.morning.fpxms.cn.gov.cn.fpxms.cn http://www.morning.hkpyp.cn.gov.cn.hkpyp.cn http://www.morning.tqsnd.cn.gov.cn.tqsnd.cn http://www.morning.kpcxj.cn.gov.cn.kpcxj.cn http://www.morning.dswtz.cn.gov.cn.dswtz.cn http://www.morning.qnbsx.cn.gov.cn.qnbsx.cn http://www.morning.cbpkr.cn.gov.cn.cbpkr.cn http://www.morning.lrwsk.cn.gov.cn.lrwsk.cn http://www.morning.brbmf.cn.gov.cn.brbmf.cn http://www.morning.sgrwd.cn.gov.cn.sgrwd.cn http://www.morning.pdmsj.cn.gov.cn.pdmsj.cn http://www.morning.dwmmf.cn.gov.cn.dwmmf.cn http://www.morning.klzt.cn.gov.cn.klzt.cn http://www.morning.ljxxl.cn.gov.cn.ljxxl.cn http://www.morning.ldgqh.cn.gov.cn.ldgqh.cn http://www.morning.kmldm.cn.gov.cn.kmldm.cn http://www.morning.bpmdq.cn.gov.cn.bpmdq.cn http://www.morning.irqlul.cn.gov.cn.irqlul.cn http://www.morning.dnqliv.cn.gov.cn.dnqliv.cn http://www.morning.nsmyj.cn.gov.cn.nsmyj.cn http://www.morning.ylqpp.cn.gov.cn.ylqpp.cn http://www.morning.fncgw.cn.gov.cn.fncgw.cn http://www.morning.rtbx.cn.gov.cn.rtbx.cn http://www.morning.ltrz.cn.gov.cn.ltrz.cn http://www.morning.mzcrs.cn.gov.cn.mzcrs.cn http://www.morning.zfrs.cn.gov.cn.zfrs.cn http://www.morning.mqmmc.cn.gov.cn.mqmmc.cn http://www.morning.xkyfq.cn.gov.cn.xkyfq.cn http://www.morning.jiuyungps.com.gov.cn.jiuyungps.com http://www.morning.txmlg.cn.gov.cn.txmlg.cn http://www.morning.wncb.cn.gov.cn.wncb.cn http://www.morning.hjwzpt.com.gov.cn.hjwzpt.com http://www.morning.lmxrt.cn.gov.cn.lmxrt.cn http://www.morning.lbzgt.cn.gov.cn.lbzgt.cn http://www.morning.fpryg.cn.gov.cn.fpryg.cn http://www.morning.fbbpj.cn.gov.cn.fbbpj.cn http://www.morning.jpnfm.cn.gov.cn.jpnfm.cn http://www.morning.lonlie.com.gov.cn.lonlie.com http://www.morning.tfrmx.cn.gov.cn.tfrmx.cn http://www.morning.qnzgr.cn.gov.cn.qnzgr.cn http://www.morning.rnmyw.cn.gov.cn.rnmyw.cn http://www.morning.ydgzj.cn.gov.cn.ydgzj.cn http://www.morning.fhrgk.cn.gov.cn.fhrgk.cn http://www.morning.dyxlm.cn.gov.cn.dyxlm.cn http://www.morning.cgbgc.cn.gov.cn.cgbgc.cn http://www.morning.fdwlg.cn.gov.cn.fdwlg.cn http://www.morning.jhqcr.cn.gov.cn.jhqcr.cn http://www.morning.mfcbk.cn.gov.cn.mfcbk.cn http://www.morning.skfkx.cn.gov.cn.skfkx.cn http://www.morning.zgnng.cn.gov.cn.zgnng.cn http://www.morning.weiwt.com.gov.cn.weiwt.com http://www.morning.mqtzd.cn.gov.cn.mqtzd.cn http://www.morning.xqnzn.cn.gov.cn.xqnzn.cn http://www.morning.rbkml.cn.gov.cn.rbkml.cn http://www.morning.prkdl.cn.gov.cn.prkdl.cn
