莫弦然资料

本文主要介绍莫弦然资料 方法和在新技术下所面对的“挑战”,方便大家深入理解莫弦然资料 过程。本文也将分享莫弦然资料 所遇到的问题和应对策略,怎么解决怎么做的问题。
通过深入本文可以理解代码原理,进行代码文档的下载,也可以查看相应 Demo 部署效果。

 

什么是消息队列?

就是生产者生产一条消息,发送到这个rabbitmq,消费者连接rabbitmq并且进行消费,生产者和消费者并需要知道对方是如何工作的,从而实现程序之间的解耦,异步和削峰,这也就是消息队列的作用。

使用的场景也有很多,比如用户支付购买之后的发送短信,增加用户积分等等,只要能将业务逻辑抽象出来,就能很好得使用它。

 

下面进入正题:

先来介绍一下基本概念和参与生命周期的各个成员。

publisher:消息生产者,负责创建消息,并发送到代理服务器(rabbitmq)

message:发送的消息,由 有效负载(payload) 和 标签 (label) 组成

exchange:交换器,负责接收消息并路由给服务器的队列

queue:消息队列,就是消息最后要去的地方。然后等待消费者取走并消费

consumer:消息消费者,与生产者对应,程序的另外一方,负责消费信息,并完成相应的业务逻辑

channel:信道,在tcp之上建立的通道,负责传送消息。队列的传输都是基于信道来完成的。

broker:消息队列服务器实体

下面来解析一下这张图,这张图是网上找的,虽然不够详细,但是勉强能用。

莫弦然

 

准备前提,开启rabbitmq服务,生命队列和交换器,并将两者进行绑定。

 

生产逻辑:

publisher 通过 broker服务器ip 和 端口 尝试建立与 broker 的 tcp 连接,连接成功之后,会尝试验证验证用户名和密码,如果错误,则拒绝访问。

验证成功之后,在tcp上建立一个 信道(channel) ,publisher 通过信道,发送消息到 broker,进入指定的虚拟主机virtual host。然后查找到交换器绑定的队列,并将消息推送进入队列。

 

消费逻辑:

consumer 建立连接和验证和生产者一样。接下来通过队列名,直接找到队列进行监听并消费。

 

有几个比较重要的知识点:

1.  rabbitmq的交换器并不是真正意义的交换器,它本质上其实就是一张表,里面存放和交换器名称和消息队列的映射关系。所以说队列的传输都是通过信道来完成的。

2.  信道存在的意义在于 创建和销毁tcp连接非常消耗资源

 

 

交换器的类型

当生产的消息进入虚拟主机时,会去寻找一张表,就是交换器和队列映射关系的一张表。而交换器的类型也分了好多种,可以根据不同的场景自由选择。

目前总共分了4种,direct,fanout,topic,headers。其中headers因为性能问题几乎不在使用,这里就不做过多的讨论。

1.direct

direct是直接,完全匹配,单播的模式。

莫弦然

php简单代码实现:

//创建队列 $q = new AMQPQueue($channel); $q->setName($q_name); $q->setFlags(AMQP_DURABLE); //持久化 $q->declare();  //创建交换机对象 $ex = new AMQPExchange($channel); $ex->setName($e_name); $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型 $ex->setFlags(AMQP_DURABLE); //持久化 $ex->declare();  //绑定交换机与队列,并指定路由键 $q->bind($e_name, $k_route); 
//生产消息
$ex->publish($message, $k_route)

特性: 当生产消息时,未指定交换器,则会默认使用 (AMQP default) 交换器,然后路由到 和路由键名称相同的队列中去

 

 

2.fanout

如果说 direct 是 单对单 的关系,那么 fanout就是单对多的关系,即一个交换器对应多个队列。

fanout交换器不通过路由键路由到队列,而是通过将队列绑定在交换器上,当消息进来时,直接路由到该交换器绑定的队列去。

莫弦然

 

 

3.topic

topic交换器通过路由键,将自动匹配允许匹配的队列,相比fanout,更加灵活,不过对架构要求更高。如下图所示

莫弦然

$e_name = 'logs-exchange'; //交换机名 $q_name = 'msg-inbox-errors'; //队列名  //创建队列 $q = new AMQPQueue($channel); $q->setName($q_name); $q->setFlags(AMQP_DURABLE); //持久化 $q->declare();  //创建交换机对象 $ex = new AMQPExchange($channel); $ex->setName($e_name); $ex->setType(AMQP_EX_TYPE_TOPIC); //direct类型 $ex->setFlags(AMQP_DURABLE); //持久化 $ex->declare();   $q->bind($e_name, '*.msg-inbox');  $ex->publish($message, 'wonima.msg-inbox');

 

高级特性:

为了确保消息可靠性,有两种处理方式.

1.rabbitmq事务

事务主要是对信道进行设置,示例代码如下

$channel->startTransaction(); //开始事务 for($i=0; $i<5; ++$i){     $message = "TEST MESSAGE! 测试消息!";     $message = $message.$i."---";     echo "Send Message:".$ex->publish($message, 'xxxx')."n";  } $channel->commitTransaction(); //提交事务

经测验,使用事务之后,性能会造成相当大的影响,与不实用事务相比,性能可以相差百倍以上。

 

2.confirm 模式

当信道设置未 confirm 模式的时候,每一条消息都会获的唯一的id。当消费者接收到消息的时候,自动发送 或 手动发送消息  进行消息确认。

//创建队列 $q = new AMQPQueue($channel); $q->setName($q_name); $q->setFlags(AMQP_DURABLE); //持久化 echo "Message Total:".$q->declare()."n";  //第一种:自动应答 //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答 
//第二种:手动应答 $q->consume('processMessage'); /** * 消费回调函数 * 处理消息 */ function processMessage($envelope, $queue) { $msg = $envelope->getBody(); sleep(2); $myfile = fopen("newfile2.txt", "a+") or die("Unable to open file!"); $txt = $msg.time()."n"; fwrite($myfile, $txt); fclose($myfile); echo $msg.time()."n"; //处理消息 $q->ack($envelope->getDeliveryTag()); //手动发送ACK应答 }

 

应答模式最大的好处是就是异步,执行效率高。事务和应答模式相比,后者使用更加频繁,前者几乎没有见到过。

 

延迟队列

首先声明rabbitmq是不支持延迟队列的,但是我们可以利用死信队列来完成。

莫弦然

 

 

实现延迟队列也有多种方式:

第一种:设置死信队列,并将 过期时间 加到队列里面

try {     $conn = new AMQPConnection($connectConfig);     $conn->connect();     if (!$conn->isConnected()) {          echo 'rabbit-mq 连接错误:', json_encode($connectConfig);         exit();     }     $channel = new AMQPChannel($conn);     if (!$channel->isConnected()) {         echo 'rabbit-mq Connection through channel failed:', json_encode($connectConfig);         exit();     }     $exchange = new AMQPExchange($channel);     $exchange->setFlags(AMQP_DURABLE);//持久化     $exchange->setName($params['exchangeName'] ?: '');     $exchange->setType(AMQP_EX_TYPE_DIRECT); //direct类型     $exchange->declareExchange();      $queue = new AMQPQueue($channel);     $queue->setName($params['queueName'] ?: '');     $queue->setFlags(AMQP_DURABLE);     $queue->setArguments(array(         'x-dead-letter-exchange' => 'last_exchange',         'x-dead-letter-routing-key' => 'last_route',         'x-message-ttl' => 10000,     ));     $queue->declareQueue();      //绑定     $queue->bind($params['exchangeName'], $params['routeKey']);      $exchange2 = new AMQPExchange($channel);     $exchange2->setFlags(AMQP_DURABLE);//持久化     $exchange2->setName('last_exchange');     $exchange2->setType(AMQP_EX_TYPE_DIRECT); //direct类型     $exchange2->declareExchange();      $queue2 = new AMQPQueue($channel);     $queue2->setName('last_queue');     $queue2->setFlags(AMQP_DURABLE);     $queue2->declareQueue();      $queue2->bind('last_exchange', 'last_queue');  } catch (Exception $e) {  }  $time = time(); //生成消息 $exchange->publish((string)$time, $params['routeKey'], AMQP_MANDATORY, [     'delivery_mode' => 2, ]);

 

第二种:设置死信队列,并将 过期时间 加到消息里面,这一种更加自由。

$msg = [     'x-message-ttl' => 5,     'ttl' => 5,     'body' => time() ]; $msg = json_encode($msg); $exchange->publish($msg, '', AMQP_MANDATORY, ['delivery_mode' => 2]);

 

第三种:使用延迟插件

 

集群

先来谈谈rabbitmq的集群是如何运行的

 

当你开启来两个rabbitmq(节点)服务,并将其组成为一个集群。每个节点并不会将所有的队列进行拷贝,元数据依旧保存在单个节点当中,其他节点则是通过指针。

举个例子:节点a和节点b组成了一个集群,节点a保存着一堆元数据 c 和 元数据d的指针,用来指向节点b,节点b保存一堆元数据d 和 元数据 c的指针,用来指向节点a。

这样做有两个原因

1 存储空间 :如果一个节点存储了1gb的数据,再添加节点,只会带来一摸一样的1gb的数据,非常浪费磁盘空间

2 性能:对于持久化消息来说,每一条消息都会触发磁盘io,每次新增节点,网路和磁盘负载都会增加,相对于单机来说,性能不但不会提升,反而可能下降。

但是由于交换器只是一张查询表,并非实际的路由器,因此将交换器在整个集群进行复制也不会损耗太多的性能,所以交换器在每个节点都会保存一份,以便于查询。

 

莫弦然资料部分资料来自网络,侵权毕设源码联系删除

区块链毕设网(www.qklbishe.com)全网最靠谱的原创区块链毕设代做网站
部分资料来自网络,侵权联系删除!
资源收费仅为搬运整理打赏费用,用户自愿支付 !
qklbishe.com区块链毕设代做网专注|以太坊fabric-计算机|java|毕业设计|代做平台 » 莫弦然资料

提供最优质的资源集合

立即查看 了解详情