网站源码模板,wordpress 点击文章,网站开发服务 税,佛山市住房和城乡建设部网站深入了解RabbitMQ工作原理及简单使用 RabbitMQ系列文章 RabbitMQ在Ubuntu上的环境搭建深入了解RabbitMQ工作原理及简单使用RabbitMQ交换器Exchange介绍与实践RabbitMQ事务和Confirm发送方消息确认——深入解读使用Docker部署RabbitMQ集群你不知道的RabbitMQ集群架构全解RabbitM… 深入了解RabbitMQ工作原理及简单使用 RabbitMQ系列文章 RabbitMQ在Ubuntu上的环境搭建深入了解RabbitMQ工作原理及简单使用RabbitMQ交换器Exchange介绍与实践RabbitMQ事务和Confirm发送方消息确认——深入解读使用Docker部署RabbitMQ集群你不知道的RabbitMQ集群架构全解RabbitMQ简介 在介绍RabbitMQ之前实现要介绍一下MQMQ是什么 MQ全称是Message Queue可以理解为消息队列的意思简单来说就是消息以管道的方式进行传递。 RabbitMQ是一个实现了AMQPAdvanced Message Queuing Protocol高级消息队列协议的消息队列服务用Erlang语言的。 使用场景 在我们秒杀抢购商品的时候系统会提醒我们稍等排队中而不是像几年前一样页面卡死或报错给用户。 像这种排队结算就用到了消息队列机制放入通道里面一个一个结算处理而不是某个时间断突然涌入大批量的查询新增把数据库给搞宕机所以RabbitMQ本质上起到的作用就是削峰填谷为业务保驾护航。 为什么选择RabbitMQ 现在的市面上有很多MQ可以选择比如ActiveMQ、ZeroMQ、Appche Qpid那问题来了为什么要选择RabbitMQ 除了QpidRabbitMQ是唯一一个实现了AMQP标准的消息服务器可靠性RabbitMQ的持久化支持保证了消息的稳定性高并发RabbitMQ使用了Erlang开发语言Erlang是为电话交换机开发的语言天生自带高并发光环和高可用特性集群部署简单正是应为Erlang使得RabbitMQ集群部署变的超级简单社区活跃度高根据网上资料来看RabbitMQ也是首选工作机制 生产者、消费者和代理 在了解消息通讯之前首先要了解3个概念生产者、消费者和代理。 生产者消息的创建者负责创建和推送数据到消息服务器 消费者消息的接收方用于处理数据和确认消息 代理就是RabbitMQ本身用于扮演“快递”的角色本身不生产消息只是扮演“快递”的角色。 消息发送原理 首先你必须连接到Rabbit才能发布和消费消息那怎么连接和发送消息的呢 你的应用程序和Rabbit Server之间会创建一个TCP连接一旦TCP打开并通过了认证认证就是你试图连接Rabbit之前发送的Rabbit服务器连接信息和用户名和密码有点像程序连接数据库使用Java有两种连接认证的方式后面代码会详细介绍一旦认证通过你的应用程序和Rabbit就创建了一条AMQP信道Channel。 信道是创建在“真实”TCP上的虚拟连接AMQP命令都是通过信道发送出去的每个信道都会有一个唯一的ID不论是发布消息订阅队列或者介绍消息都是通过信道完成的。 为什么不通过TCP直接发送命令 对于操作系统来说创建和销毁TCP会话是非常昂贵的开销假设高峰期每秒有成千上万条连接每个连接都要创建一条TCP会话这就造成了TCP连接的巨大浪费而且操作系统每秒能创建的TCP也是有限的因此很快就会遇到系统瓶颈。 如果我们每个请求都使用一条TCP连接既满足了性能的需要又能确保每个连接的私密性这就是引入信道概念的原因。 你必须知道的Rabbit 想要真正的了解Rabbit有些名词是你必须知道的。 包括ConnectionFactory连接管理器、Channel信道、Exchange交换器、Queue队列、RoutingKey路由键、BindingKey绑定键。 ConnectionFactory连接管理器应用程序与Rabbit之间建立连接的管理器程序代码中使用 Channel信道消息推送使用的通道 Exchange交换器用于接受、分配消息 Queue队列用于存储生产者的消息 RoutingKey路由键用于把生成者的数据分配到交换器上 BindingKey绑定键用于把交换器的消息绑定到队列上 看到上面的解释最难理解的路由键和绑定键了那么他们具体怎么发挥作用的请看下图 关于更多交换器的信息我们在后面再讲。 消息持久化 Rabbit队列和交换器有一个不可告人的秘密就是默认情况下重启服务器会导致消息丢失那么怎么保证Rabbit在重启的时候不丢失呢答案就是消息持久化。 当你把消息发送到Rabbit服务器的时候你需要选择你是否要进行持久化但这并不能保证Rabbit能从崩溃中恢复想要Rabbit消息能恢复必须满足3个条件 投递消息的时候durable设置为true消息持久化代码channel.queueDeclare(x, true, false, false, null)参数2设置为true持久化设置投递模式deliveryMode设置为2持久代码channel.basicPublish(x, x, MessageProperties.PERSISTENT_TEXT_PLAIN,x)参数3设置为存储纯文本到磁盘消息已经到达持久化交换器上消息已经到达持久化的队列持久化工作原理 Rabbit会将你的持久化消息写入磁盘上的持久化日志文件等消息被消费之后Rabbit会把这条消息标识为等待垃圾回收。 持久化的缺点 消息持久化的优点显而易见但缺点也很明显那就是性能因为要写入硬盘要比写入内存性能较低很多从而降低了服务器的吞吐量尽管使用SSD硬盘可以使事情得到缓解但他仍然吸干了Rabbit的性能当消息成千上万条要写入磁盘的时候性能是很低的。 所以使用者要根据自己的情况选择适合自己的方式。 虚拟主机 每个Rabbit都能创建很多vhost我们称之为虚拟主机每个虚拟主机其实都是mini版的RabbitMQ拥有自己的队列交换器和绑定拥有自己的权限机制。 vhost特性 RabbitMQ默认的vhost是“/”开箱即用 多个vhost是隔离的多个vhost无法通讯并且不用担心命名冲突队列和交换器和绑定实现了多层分离 创建用户的时候必须指定vhost vhost操作 可以通过rabbitmqctl工具命令创建 rabbitmqctl add_vhost[vhost_name] 删除vhost rabbitmqctl delete_vhost[vhost_name] 查看所有的vhost rabbitmqctl list_vhosts 环境搭建 前文我们已经介绍了Ubuntu搭建RabbitMQ的步骤RabbitMQ在Ubuntu上的环境搭建 如果你是在Windows10上去安装那就更简单了先放下载地址 Erlang/Rabbit Server百度网盘链接https://pan.baidu.com/s/1TnKDV-ZuXLiIgyK8c8f9dg 密码wct9 当然也可去Erlang和Rabbit官网去下就是速度比较慢。我的百度云Rabbit最新版本3.7.6Erlang版本20.2注意不要下载最新的Erlang在Windows10上打开扩展插件有问题打不开。 安装Erlang 安装Rabbit Server 进入安装目录\sbin下使用命令“rabbitmq-plugins enable rabbitmq_management”启动网页管理插件 重启Rabbit服务 使用http://localhost:15672进行测试默认的登陆账号为guest密码为guest 重复安装Rabbit Server的坑 如果不是第一次在Windows上安装Rabbit Server一定要把Rabbit和Erlang卸载干净之后找到注册表HKEY_LOCAL_MACHINE\SOFTWARE\Ericsson\Erlang\ErlSrv 删除其下的所有项。 不然会出现Rabbit安装之后启动不了的情况理论上卸载的顺序也是先Rabbit在Erlang。 代码实现 java版实现使用maven项目创建可以查看MyEclipse2017破解设置与maven项目搭建 项目创建成功之后添加Rabbit Client jar包只需要在pom.xml里面配置如下信息 dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.2.0/version
/dependency java实现代码分为两个类第一个是创建Rabbit连接第二是应用类使用最简单的方式发布和消费消息。 Rabbit的连接两种方式 方式一 public static Connection GetRabbitConnection() {ConnectionFactory factory new ConnectionFactory();factory.setUsername(Config.UserName);factory.setPassword(Config.Password);factory.setVirtualHost(Config.VHost);factory.setHost(Config.Host);factory.setPort(Config.Port);Connection conn null;try {conn factory.newConnection();} catch (Exception e) {e.printStackTrace();}return conn;
} 方式二 public static Connection GetRabbitConnection2() {ConnectionFactory factory new ConnectionFactory();// 连接格式amqp://userName:passwordhostName:portNumber/virtualHostString uri String.format(amqp://%s:%s%s:%d%s, Config.UserName, Config.Password, Config.Host, Config.Port,Config.VHost);Connection conn null;try {factory.setUri(uri);factory.setVirtualHost(Config.VHost);conn factory.newConnection();} catch (Exception e) {e.printStackTrace();}return conn;
} 第二部分应用类使用最简单的方式发布和消费消息 public static void main(String[] args) {Publisher(); // 推送消息Consumer(); // 消费消息
}/*** 推送消息*/
public static void Publisher() {// 创建一个连接Connection conn ConnectionFactoryUtil.GetRabbitConnection();if (conn ! null) {try {// 创建通道Channel channel conn.createChannel();// 声明队列【参数说明参数一队列名称参数二是否持久化参数三是否独占模式参数四消费者断开连接时是否删除队列参数五消息其他参数】channel.queueDeclare(Config.QueueName, false, false, false, null);String content String.format(当前时间%s, new Date().getTime());// 发送内容【参数说明参数一交换机名称参数二队列名称参数三消息的其他属性-routing headers此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘参数四消息主体】channel.basicPublish(, Config.QueueName, null, content.getBytes(UTF-8));System.out.println(已发送消息 content);// 关闭连接channel.close();conn.close();} catch (Exception e) {e.printStackTrace();}}
}/*** 消费消息*/
public static void Consumer() {// 创建一个连接Connection conn ConnectionFactoryUtil.GetRabbitConnection();if (conn ! null) {try {// 创建通道Channel channel conn.createChannel();// 声明队列【参数说明参数一队列名称参数二是否持久化参数三是否独占模式参数四消费者断开连接时是否删除队列参数五消息其他参数】channel.queueDeclare(Config.QueueName, false, false, false, null);// 创建订阅器并接受消息channel.basicConsume(Config.QueueName, false, , new DefaultConsumer(channel) {Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String routingKey envelope.getRoutingKey(); // 队列名称String contentType properties.getContentType(); // 内容类型String content new String(body, utf-8); // 消息正文System.out.println(消息正文 content);channel.basicAck(envelope.getDeliveryTag(), false); // 手动确认消息【参数说明参数一该消息的index参数二是否批量应答true批量确认小于index的消息】}});} catch (Exception e) {e.printStackTrace();}}
} 代码里面已经写了很详细的注释在这里也不过多的介绍了。 执行效果如图 转载于:https://www.cnblogs.com/javaGoGo/p/10111513.html
