队列

app-ghs/controllers/TestController.php

@@ -39,7 +39,14 @@ use wkhtmltox\Image\Converter;
 class TestController extends BaseController
 {
 
-    public $guestAccess = ['recharge', 'fx', 'on', 'in', 'clear-query', 'ip', 'html', 'debt', 'apply', 'clear', 'ls-trade-query', 'ls-trade-refund', 'refund-query', 'inform', 'notice', 'order-query', 'add-order', 'index', 'get', 'trade-query', 'query', 'balance-query', 'refund', 'recharge-query', 'sim', 'push', 'code'];
+    public $guestAccess = ['rabbit','recharge', 'fx', 'on', 'in', 'clear-query', 'ip', 'html', 'debt', 'apply', 'clear', 'ls-trade-query', 'ls-trade-refund', 'refund-query', 'inform', 'notice', 'order-query', 'add-order', 'index', 'get', 'trade-query', 'query', 'balance-query', 'refund', 'recharge-query', 'sim', 'push', 'code'];
+
+    public function actionRabbit()
+    {
+        $producer = Yii::$app->rabbitmq->getProducer('producers');
+        $msg = serialize(['user_id' => rand(111,999)]);
+        $producer->publish($msg, 'ex1', 'route1');
+    }
 
     public function actionFx()
     {

common/config/rabbitMQ.php

@@ -31,37 +31,29 @@ $rabbitMQ = [
     /**
      * 交换器
      * 交换类型(type)：direct、topic、headers、fanout
-     * durable => true: 交换器持久化，即使服务器重启也不会丢失
      * 列出交换器: rabbitmqctl list_exchanges
      */
     'exchanges' => [
         [
-            'name' => 'customExchange',
-            'type' => 'direct',
-            'passive' => false,
-            'durable' => true,
-            'auto_delete' => false,
-            'internal' => false,
-            'nowait' => false,
-            'arguments' => null,
-            'ticket' => null,
-            'declare' => true,
+            'name' => 'commonExchange',
+            'type' => 'direct'
         ],
         [
-            'name' => 'stockExchange',
-            'type' => 'direct',
-            'passive' => false,
-            'durable' => true,
-            'auto_delete' => false,
-            'internal' => false,
-            'nowait' => false,
-            'arguments' => null,
-            'ticket' => null,
-            'declare' => true,
+            'name' => 'informExchange',
+            'type' => 'direct'
+        ],
+        [
+            'name' => 'openShopInitExchange',
+            'type' => 'direct'
         ],
         [
-            'name' => 'notifyExchange',
-            'type' => 'direct',
+            'name' => 'ex1',
+            'type' => 'direct'
+        ],
+        /*
+        [
+            'name' => 'topic_ex_1',
+            'type' => 'topic', // topic
             'passive' => false,
             'durable' => true,
             'auto_delete' => false,
@@ -71,27 +63,32 @@ $rabbitMQ = [
             'ticket' => null,
             'declare' => true,
         ],
+        */
     ],
 
     /**
      * 队列
-     * durable => true: 队列持久化，即使服务器重启也不会丢失
-     * auto_delete => false: 队列在没有消费者时不会自动删除
      */
     'queues' => [
         [
-            'name' => 'customQueue',
+            'name' => 'queues1',
+            'passive' => false,
+        ],
+        [
+            'name' => 'queueCommon',
+            'passive' => false,
+        ],
+        [
+            'name' => 'informQueue',
             'passive' => false,
-            'durable' => true,
-            'exclusive' => false,
-            'auto_delete' => false,
-            'nowait' => false,
-            'arguments' => null,
-            'ticket' => null,
-            'declare' => true,
         ],
         [
-            'name' => 'stockQueue',
+            'name' => 'openShopInitQueue',
+            'passive' => false,
+        ],
+        /*
+        [
+            'name' => 'queue_2',
             'passive' => false,
             'durable' => true,
             'exclusive' => false,
@@ -102,7 +99,7 @@ $rabbitMQ = [
             'declare' => true,
         ],
         [
-            'name' => 'notifyQueue',
+            'name' => 'queue_3',
             'passive' => false,
             'durable' => true,
             'exclusive' => false,
@@ -112,6 +109,7 @@ $rabbitMQ = [
             'ticket' => null,
             'declare' => true,
         ],
+        */
     ],
 
     /**
@@ -120,115 +118,119 @@ $rabbitMQ = [
      */
     'bindings' => [
         [
-            'queue' => 'customQueue',
-            'exchange' => 'customExchange',
-            'routing_keys' => ['customRoute'],
+            'queue' => 'queueCommon',
+            'exchange' => 'commonExchange',
+            'routing_keys' => ['commonRoute'],
         ],
         [
-            'queue' => 'stockQueue',
-            'exchange' => 'stockExchange',
-            'routing_keys' => ['stockRoute'],
+            'queue' => 'informQueue',
+            'exchange' => 'informExchange',
+            'routing_keys' => ['informRoute'],
         ],
         [
-            'queue' => 'notifyQueue',
-            'exchange' => 'notifyExchange',
-            'routing_keys' => ['notifyRoute'],
+            'queue' => 'openShopInitQueue',
+            'exchange' => 'openShopInitExchange',
+            'routing_keys' => ['openShopInitRoute'],
         ],
+        [
+            'queue' => 'queues1',
+            'exchange' => 'ex1',
+            'routing_keys' => ['route1'],
+        ],
+        /*
+        [
+            'queue' => 'queue_2',
+            'exchange' => 'topic_ex_1',
+            'routing_keys' => ['route_1', 'route_prefix.*', '*.route_suffix', '*.*.last', '*'],
+        ],
+        [
+            'queue' => 'queue_3',
+            'exchange' => 'topic_ex_1',
+            'routing_keys' => ['route_3', 'route3_prefix.*', '*.route_suffix', '*.*.last'],
+        ],
+        */
     ],
 
     /**
      * 生产者
-     * delivery_mode => 2: 消息持久化，即使服务器重启也不会丢失
-     * safe => true: 确保消息被正确发送
      */
     'producers' => [
         [
-            //客户操作生产者
-            'name' => 'customProducer',
-            'connection' => 'default',
-            'safe' => true,
-            'content_type' => 'application/json',
-            'delivery_mode' => 2,
-            'serializer' => 'json_encode',
+            //通用生产者
+            'name' => 'commonProducer',
         ],
         [
-            //库存管理生产者
-            'name' => 'stockProducer',
-            'connection' => 'default',
-            'safe' => true,
-            'content_type' => 'application/json',
-            'delivery_mode' => 2,
-            'serializer' => 'json_encode',
+            //通知生产者
+            'name' => 'informProducer',
         ],
         [
             //通知生产者
-            'name' => 'notifyProducer',
+            'name' => 'openShopInitProducer',
+        ],
+        [
+            'name' => 'producers',
+        ],
+        /*
+        [
+            'name' => 'producer_2',
             'connection' => 'default',
             'safe' => true,
-            'content_type' => 'application/json',
+            'content_type' => 'text/plain',
             'delivery_mode' => 2,
-            'serializer' => 'json_encode',
-        ],
+            'serializer' => 'serialize',
+        ]
+        */
     ],
 
     /**
      * 消费者
-     * qos.prefetch_count => 1: 保证只有一条消息未确认时才会收到下一条，防止消息丢失
-     * proceed_on_exception => true: 消费异常时继续处理下一条消息
-     * deserializer => 'json_decode': 消息反序列化
      */
     'consumers' => [
         [
-            //客户操作消费者
-            'name' => 'customConsumer',
-            'connection' => 'default',
+            //常规消费者
+            'name' => 'commonConsumer',
             'callbacks' => [
-                'customQueue' => '\common\components\rabbitmq\consumer\customConsumer',
+                'queueCommon' => '\common\components\rabbitmq\consumer\commonConsumer', //设定队列的数据给谁转回调而以，只是个配置
             ],
-            'qos' => [
-                'prefetch_size' => 0,
-                'prefetch_count' => 1,
-                'global' => false,
+        ],
+        [
+            //通知
+            'name' => 'informConsumer',
+            'callbacks' => [
+                'informQueue' => '\common\components\rabbitmq\consumer\informConsumer', //设定队列的数据给谁转回调而以，只是个配置
             ],
-            'idle_timeout' => 0,
-            'idle_timeout_exit_code' => null,
-            'proceed_on_exception' => true,
-            'deserializer' => 'json_decode',
         ],
         [
-            //库存管理消费者
-            'name' => 'stockConsumer',
-            'connection' => 'default',
+            //授权公众号开店初始化
+            'name' => 'openShopInitConsumer',
             'callbacks' => [
-                'stockQueue' => '\common\components\rabbitmq\consumer\stockConsumer',
+                'openShopInitQueue' => '\common\components\rabbitmq\consumer\openShopInitConsumer', //设定队列的数据给谁转回调而以，只是个配置
             ],
-            'qos' => [
-                'prefetch_size' => 0,
-                'prefetch_count' => 1,
-                'global' => false,
+        ],
+        [
+            'name' => 'consumers',
+            'callbacks' => [
+                'queues1' => '\common\components\rabbitmq\ImportDataConsumer',
             ],
-            'idle_timeout' => 0,
-            'idle_timeout_exit_code' => null,
-            'proceed_on_exception' => true,
-            'deserializer' => 'json_decode',
         ],
+        /*
         [
-            //通知消费者
-            'name' => 'notifyConsumer',
+            'name' => 'consumer_2',
             'connection' => 'default',
             'callbacks' => [
-                'notifyQueue' => '\common\components\rabbitmq\consumer\notifyConsumer',
+                'queue_2' => '\common\components\rabbitmq\consumer\PrintMsgConsumer',
             ],
             'qos' => [
                 'prefetch_size' => 0,
-                'prefetch_count' => 1,
+                'prefetch_count' => 0,
                 'global' => false,
             ],
             'idle_timeout' => 0,
             'idle_timeout_exit_code' => null,
-            'proceed_on_exception' => true,
-            'deserializer' => 'json_decode',
+            'proceed_on_exception' => false,
+            'deserializer' => 'unserialize',
         ],
+        */
     ],
 ];