网络协议。
基于TCP上面架构更高层次的功能框架。
这里主要是异步,中间服务器,多个客户端角色。多对多的情形。
发布--订阅模式
Mqtt—message queueing telemetry transport
发布者 (Publish)、代理 (Broker)(服务器)、订阅者 (Subscribe)
就是消息分了类型,然后指定某个类型接收
队列模式--功能更加复杂
amqp-advanced message queueing protocol
有各种路由策略
RabbitMQ
即可以用amqp,也可以基于mqtt。又抽象了一层结构。
Producer:消息的发布者
Consumer:消息的接收者
安装
大致先装Erlang环境,设置环境参数
装RabbitMQ。为了更直观显示状态。装managment插件。开server服务。就可以网页访问
用户名密码-guest
管理网站-ip:15672
服务器的管理密码和配置里的连接密码一样的?
连接状态基本都能看清楚。看着C/E/Q里都是没有数据交互的?
exchange-amp.xx-感觉是系统自建的。
mqtt
服务端-RabbitMQ装mqtt插件。
https://www.rabbitmq.com/mqtt.html
测试客户端-MQTTX
协议文档
http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html
https://www.cnblogs.com/schips/p/12272372.html
管理
https://www.cxyzjd.com/article/guo1wu3shi4/94861839
说是插件件基于exchanges 和 queues实现
exchange (默认为amq.topic)
默认情况下,如果Queue不存在,会自动创建一个Auto-Delete类型的队列。
没看到这个队列。。
java实现
一般是用Eclipse Paho Java Client包
https://www.eclipse.org/paho/index.php?page=clients/java/index.php
里面有例子。
源码解析https://www.jianshu.com/p/5d8d58c081e5
流程结构
client信息
clientId
认证信息
MqttConnectOptions
broker信息
message
topic--消息类别
payload--
qos--发送质量
Retained --是否保存最后一条message
连接
设置
发布
setmessage
Qos
Payload
回调
订阅-设置主题,
等待回调
topic--有通配符 ,有层级,主要靠这个来适配需求
amqp
官方的教程
https://www.rabbitmq.com/tutorials/tutorial-one-java.html
翻译https://rabbitmq.mr-ping.com/tutorials_with_csharp/rpc.html
java实现包
https://www.rabbitmq.com/api-guide.html
api列表https://rabbitmq.github.io/rabbitmq-java-client/api/current/
后端用的是com.rabbitmq.client
中间件用的是 org.springframework.amqp.rabbit
测试
网页里打开一个queue
下面有send 和 get功能。
这里一个连接里生产者和消费者是只能有一个的?
要发送和接受就要连两路服务器。?
RPC模式里是既能发布,又能消费的。
在多个环节又扩充成集合对象。每个对象之间都是多对多关系。排列组合在每个环节都倍增。
时常都想不起怎么对应的
看着这个模型用于收费模式的搭建也挺合适。
client和server的连接可以有多个channel。
server分派消息有多种路由(routes)规则。
每种规则是一个交换机(exchange),有各自的绑定队列(queue)集合。
一个队列又能绑定到多个exchange?
队列和交换机之间有个绑定操作。
交换机已经分规则了。交换机到队列之间的规则能再次定义。。?
应该是定义细则,如直连交换机,可以绑定多个键值
队列,交换机和绑定统称为AMQP实体(AMQP entities).
AMQP的实体和路由规则这些是可以应用自定义的。。
channel--在Tcp连接中扩展出多路逻辑信道,相当于多个独立连接。
Virtual Host--在服务器上扩展出多个虚拟主机?
多种交换(路由)模式--exchange
-
direct--单播/直连交换。消息发送到特定键值的队列。
同时可以多个队列申明同一个键值。相当于多播。
同时一个对队列可以申明多个键值。 -
fanout--广播/扇形交换 消息发送所有绑定队列
-
topic--多播/主题交换
direct和topic都是靠routingKey来路由。
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
更细致的分配规则-均衡分配,topic类型中是否适用?
常用的实现模式
- 队列模式
多个消费者来分别处理队列消息.
这应该是属于direct交换模式。
分派规则,- 轮流平均分派。
- 均衡分派。设置一个消费者正在负载的消息数量。
![](/i/l/?n=23&i=blog/1483330/202305/1483330-20230516175742430-571257658.png)
- RPC模式
看着好像是有个专门的返回队列,然后里面有个CorrelationId可以把请求和返回联系起来。
这里说是有两种方式,- 调用时建一个临时queue。
- 用一个专门的返回queue。
这里的消费不太一样。
一般的channel.BasicConsume()设置一次就可,然后就会保持自动回调。
RPC这里Call(string message){}中每次都要 channel.BasicConsume()。
也许只是冗余。后面调用的会过滤掉??
流程
发布方
create connecetion
建channel
connection.CreateModel
建交换机
也有内置的默认交换机。
channel.ExchangeDeclare("logs", "fanout");
建队列
channel->QueueDeclare()
绑定队列到交换机
channel.QueueBind(queue: queueName,exchange: "logs", routingKey: "");
有些地方没绑定动作。。
是默认绑定到默认路由?
发送消息
channel->Publish
消费方
建一个消费对象
consumer = new EventingBasicConsumer(channel)
写处理函数。。
绑定消费对象和队列
channel.BasicConsume(queue: "hello", autoAck: true, consumer: consumer);
spring.amqp
rabbitadmin--维护exchange,queue,binding这些
tip:
EQB--(Exchange+Queue+Bind)
EQB是共享,并不属于创建者,虽然创建的时候是某个channel发起的。
客户端分两个角色,谁来创建MQ并没约束。
感觉一般发送方是制定exchange规则,基础的Queue。并绑定。
消费方直接消费queue,或者再建queue。
看了部署文档,是事先通过管理网站先建好的。
但代码里看着各种情况都有
RabbitPlugin
exchange | queue | bind-routingkey | operate | blank |
---|---|---|---|---|
EA-MagicCommand-DIRECT | QA_MagicCommand | EA_QA(send) | basicConsume(QA) | 消费方建exchange |
EB-command | commandSend commnadResult |
send result |
RabbitPlugin.send时调用,没看到建exchange。 | |
EC-NBCommand | NBCommand | RPC时用的? | ||
data | receiveData | 当前应用没涉及到? |
RPC调用
exchange只是个名字,并没有创建过程。
是被调用端创建么?
被调用端一般只管绑定队列的。不涉及到exchange。
感觉应该是要先建好exchange。这里只是传名字。
机制
消息确认--怎么确认,未确认会怎么处理。
消息持久化--rabbitMq崩溃也还在。
公平调度
由于灵活配置,
同时各种对应关系多,不容易看清。
标签:exchange,--,绑定,RabbitMQ,queue,队列,消息中间件,channel From: https://www.cnblogs.com/halfwake/p/17406422.html