做网站被坑,做企业网站找谁,深圳网络有限公司有哪些,变性WordPressRabbitMQ 工作机制图#xff1a;
Connection#xff1a; 代表客户端#xff08;包括消息生产者和消费者#xff09;与RabbitMQ之间的连接。 Channel#xff1a; 连接内部的Channel。channel#xff1a;通道 Exchange#xff1a; 充当消息交换机的组件。 Queue#xff…RabbitMQ 工作机制图
Connection 代表客户端包括消息生产者和消费者与RabbitMQ之间的连接。 Channel 连接内部的Channel。channel通道 Exchange 充当消息交换机的组件。 Queue 消息队列。
★ 消费消息
使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下
1创建ConnectionFactory设置连接信息再通过ConnectionFactory获取Connection。
2通过Connection获取Channel。
3根据需要调用Channel的queueDeclare()方法声明队列如果声明的队列已存在该方法直接获取已有的队列如果声明的队列还不存在该方法将会创建新的队列。
4调用Channel 的 basicConsume()方法开始处理消息调用该方法时需要传入一个Consumer参数 该参数相当于JMS中的消息监听器。
这个 basicConsume()方法 相当于是异步消费。 而同步消费会出现阻塞情况这就失去消息中间件存在的意义所以先讲异步消费。
★ 发送消息
使用RabbitMQ Java Client依赖库开发消息生产者的大致步骤如下
1创建ConnectionFactory设置连接信息再通过ConnectionFactory获取Connection。
2通过Connection获取Channel。
3根据需要调用exchangeDeclare()、queueDeclare()方法声明Exchange和队列、并完成队列与Exchange的绑定。 如果声明的Exchange还不存在则创建该Exchange否则直接使用已有的Exchange。 Declare:声明、宣布
4调用Channel的basicPublish()方法发送消息调用该方法的第一个参数是exchange 第二个参数为路由key最后两个参数依次是消息属性和消息数据体。
【注意】虽然消息生产者与队列是完全隔离的 但如果消息生产者不声明消息队列那系统中就可能暂时还没有任何消息队列。
在这种情况下消息生产者向Exchange发送的消息将不会分发给任何队列这些消息直接就被丢弃了。
【备注】为了保证消息生产者能将消息发送到指定队列消息生产者需要声明消息队列保证消息队列的存在。
**问题**消息生产者 和 消息队列 是完全隔离的但是生产者为什么还要声明消息队列 **原因**因为程序如果先运行消息生产者后运行消费者而声明消息队列的方法又只存在消费者那边那么在先运行消息生产者时就会因为还没有生成消息队列所以生产者发送到exchange的消息会因为没有对应的消息队列而被丢弃。
代码演示
先创建一个普通的 maven 项目。 添加一些属性 和 RabbitMQ的依赖 创建消息消费者
把创建连接的代码封装到一个方法里去。
消费者的代码
注意channel.basicConsume 的第二个参数 autoAcktrue就是表示自动确认消息已经被消费完成了。就是当消费者接收到消息之后就立马返回一个已经确认消费的消息回去给消息队列。 这样容易出现问题就是消费者这边因为一收到消息就会自动确认消息被消费了并返回已经消费消息的结果回去给消息队列但是可能消费者其实还没有把消息消费掉而消息队列那边又以为消费者已经把消息消费了所以就继续发消息给那个消费者。 而消费者一收到消息又自动确认消费并返回就会导致这个消息队列的消息越来越多然后消费者消费不完。 这个演示已消费未确认的演示放最后
执行消费者 控制台查看 原本没有这个消息队列通过调用Channel的queueDeclare()方法声明队列如果声明的队列已存在该方法直接获取已有的队列如果声明的队列还不存在该方法将会创建新的队列
一开始消费者声明的这个消息队列这个是否独占的exclusive 参数我是写true 所以下图的 myQueue01的 Features 就是 Excl 这个就是创建的消费者。 用于消费这个 myQueue01 消息队列的消费者。
后面把 exclusive 改成了 false是因为后面的生产者需要也声明这个 myQueue01 消息队列而如果这个消息队列是 独占的就没法声明了所以改成 false
创建消息生产者
生产者发送完消息就会关闭资源 消费者则是一直启动着
测试
先启动消费者或者启动生产者都一样因为生产者和消费者都有调用queueDeclare() 方法声明消息队列所以不存在发送消息后没找到对应的消息队列而导致消息被丢弃的情况。
启动消费者
然后启动生产者 生产者发送消息 再看消费者已经消费了一条消息了。 因为先启动消费者所以生产者发送的消息马上被消费了在控制台的队列就看不到了。
再测试
先启动生产者 关闭消费者然后启动生产者发送消息 可以看出消息已经生产发送到消息队列了
这一步的流程图
启动消费者消费消息
流程图
已消费未确认
注意channel.basicConsume 的第二个参数 autoAcktrue就是表示自动确认消息已经被消费完成了。就是当消费者接收到消息之后就立马返回一个已经确认消费的消息回去给消息队列。 这样容易出现问题就是消费者这边因为一收到消息就会自动确认消息被消费了并返回已经消费消息的结果回去给消息队列但是可能消费者其实还没有把消息消费掉而消息队列那边又以为消费者已经把消息消费了所以就继续发消息给那个消费者。 而消费者一收到消息又自动确认消费并返回就会导致这个消息队列的消息越来越多然后消费者消费不完。
如图因为 autoAck 为false 所以消费者消费消息后没有进行确认。这里的 unacked 条数就为1.
如果改成 autoAck 为false 那么消费者消费消息的代码要加上确认消息的方法。 这个就是手动确认消息。
完整代码
ConnectionUtil 连接工具类 package cn.ljh.app.rabbitmq.util;import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;//连接工具
public class ConnectionUtil
{//获取连接的方法public static Connection getConnection() throws IOException, TimeoutException{//创建连接工厂----这个ConnectionFactory源码可以看出有构造器所以直接new一个出来ConnectionFactory connectionFactory new ConnectionFactory();//设置连接信息connectionFactory.setHost(localhost);connectionFactory.setPort(5672);connectionFactory.setUsername(ljh);connectionFactory.setPassword(123456);connectionFactory.setVirtualHost(/); //连接虚拟主机//从连接工厂获取连接Connection connection connectionFactory.newConnection();//返回连接return connection;}
}
P2PProducer 生产者
package cn.ljh.app.rabbitmq.producer;import cn.ljh.app.rabbitmq.consumer.P2PConsumer;
import cn.ljh.app.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;//消息生产者--使用默认的exchange
public class P2PProducer
{//1创建ConnectionFactory设置连接信息再通过ConnectionFactory获取Connection。//2通过Connection获取Channel。//3根据需要调用exchangeDeclare()、queueDeclare()方法声明Exchange和队列、并完成队列与Exchange的绑定。// 如果声明的Exchange还不存在则创建该Exchange否则直接使用已有的Exchange。//4调用Channel的basicPublish()方法发送消息调用该方法的第一个参数是exchange// 第二个参数为路由key最后两个参数依次是消息属性和消息数据体。public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接Connection conn ConnectionUtil.getConnection();//2、通过Connection获取Channel。Channel channel conn.createChannel();//3、调用exchangeDeclare()方法声明Exchange、调用queueDeclare()方法声明队列并完成队列与Exchange的绑定//此处打算直接使用默认的Exchange来分发消息因此无需声明 Exchange只需声明队列channel.queueDeclare(P2PConsumer.QUEUE_NAME, true, false, false, null);String message 生产者发送的消息的内容;//4、调用Channel的basicPublish()方法发送消息channel.basicPublish(/*默认的 Exchange 没有名字所以用空的字符串*/,P2PConsumer.QUEUE_NAME/*使用队列名作为路由key表明该消息将会被路由到该队列*/,null /*指定额外的消息的属性*/,message.getBytes(StandardCharsets.UTF_8)/*消息体必须是字节数组类型--byte[]*/);//5、关闭资源//关闭通道channel.close();//关闭连接conn.close();}
}P2PConsumer 消费者
package cn.ljh.app.rabbitmq.consumer;import cn.ljh.app.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;//消息消费者
public class P2PConsumer
{// 使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下//1创建ConnectionFactory连接工厂设置连接信息再通过ConnectionFactory获取Connection连接。//2通过Connection获取Channel。//3根据需要、调用Channel的queueDeclare()方法声明队列 Declare:声明、宣布// 如果声明的队列已存在该方法直接获取已有的队列如果声明的队列还不存在该方法将会创建新的队列。//4调用Channel 的 basicConsume()方法开始处理消息调用该方法时需要传入一个Consumer参数该参数相当于JMS中的消息监听器。//常量public final static String QUEUE_NAME myQueue01;public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接工厂设置连接信息然后再通过连接工厂获取连接Connection conn ConnectionUtil.getConnection();//2、通过Connection获取Channel 消息通道Channel channel conn.createChannel();//3、调用 Channel 的 queueDeclare() 方法声明队列//如果声明的队列已存在该方法直接获取已有的队列如果声明的队列还不存在该方法将会创建新的队列//参数1声明的队列名 参数2消息队列是否持久化//参数3是否只允许该消息消费者消费该队列的消息为true则其他消费者在这个myQueue01队列消息积堆过多的情况下也无法帮忙消费。//参数4是否自动删除如果为true在该队列没消息的情况下会自动删除该队列 参数5填写额外的参数channel.queueDeclare(QUEUE_NAME, true, false, false, null);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(QUEUE_NAME/*消费这个名字的消费队列里面的消息*/,true/*消息的确认模式是否自动确认*/,new DefaultConsumer(channel){//处理消息当这个消息队列收到消息的时候这个方法就会被触发。重写这个方法Overridepublic void handleDelivery(String consumerTag,Envelope envelope /*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties properties /*消息的那些属性*/,byte[] body /*body消息的消息体*/) throws IOException{//把消息体中的消息拿出来String message new String(body, UTF-8);//printf格式化输出函数 %s输出字符串 %n换行System.err.printf(P2PConsumer收到来自Exchange为【%s】、路由key为【%s】的消息消息内容为%s%n,envelope.getExchange(),envelope.getRoutingKey(),message);}});}}
pom.xml
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcn.ljh/groupIdartifactIdrabbitmqtest/artifactIdversion1.0.0/versionnamerabbitmqtest/name!-- 属性 --propertiesmaven.compiler.source11/maven.compiler.sourcemaven.compiler.target11/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncodingjava.version11/java.version/properties!-- 依赖 --dependencies!-- RabbitMQ 的依赖库 --dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.13.0/version/dependency/dependencies/project
