关于swoft-amqp的一些简单应用

文章目录

- - 前言
  - 手动下载
  - composer.json配置
  - bean.php配置示例
  - Direct模式
  - - consumer
    - 1-1-1（exchange、route、queue数量关系）
    - - bean配置（省略其他配置）
      - productor
    - 1-1-N
    - - bean配置（省略其他配置）
      - productor
    - 1-N-N
    - - bean配置（省略其他配置）
      - productor
    - N-1-1
    - - bean配置（省略其他配置）
      - productor
    - N-N-1
    - - bean配置（省略其他配置）
      - productor

前言

参考文章：swoft使用rabbitmq消息队列

官方提供了swoft/amqp的组件，但是在package中查找不到，需要手动加入composer.json
没有文档，没有示例
2年未更新，仍然留存有使用问题

手动下载

swoft/amqp

composer.json配置

添加以下内容

"require": {
    "swoft/amqp": "^2.0"
},
{
    "repositories": {
        "swoft-amqp": {
            "type": "git",
              "url": "https://github.com/swoft-cloud/swoft-amqp.git"
        }
    }
}

bean.php配置示例

'amqp'              => [
    'class'    => SwoftAmqpClient::class,
    'host'     => 'rabbitmq',
    'port'     => 5672,
    'user' => 'guest',
    'password' => 'guest',
    'vhost' => '/',
    // 默认交换机、队列、路由名称，非强制，建议配置。当下面的channels没有具体设置时，会使用此配置
    'queue' => 'test_queue',
    'exchange' => 'exchange_test',
    'route' => 'example-test-routing-key',
    // 交换机类型。
    'type' => AMQPExchangeType::DIRECT,
    'channels' => [
        'channel_1'=>[  //对应的key即为channel_id
            'exchange' => 'exchange_test_02',
            'queue' => 'subway',
            'route' => 'route_subway',
        ],
    ],
],
'amqp.pool' => [
    'class' => SwoftAmqpPool::class,
    'client' => bean('amqp'),
],

Direct模式 consumer

不同进程消费不同队列的消息，修改channel_1为对应的信道名称即可

CLog::info('worker-' . $workerId);
CLog::info('run method: ' . __METHOD__);
// 实例化不同信道
$channel = Amqp::channel('channel_1');
$channel->listen(function ($message) {
    //$message：数据结构(json_encode)之后
    //{"body":"hey!-----9","body_size":10,"is_truncated":false,"content_encoding":null,"delivery_info":{"channel":{"callbacks":{"amq.ctag-epGZgfHej3YrjZk2FBvp0A":{}}},"delivery_tag":99,"redelivered":false,"exchange":"exchange_test","routing_key":"example-test-routing-key","consumer_tag":"amq.ctag-epGZgfHej3YrjZk2FBvp0A"}}
    CLog::info('message:' . json_encode($message));
    // 模拟业务处理
    sleep(random_int(10,20));
});

1-1-1（exchange、route、queue数量关系） bean配置（省略其他配置）

'type' => AMQPExchangeType::DIRECT,
'channels' => [
    'channel_1'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_01',
        'queue' => 'que_01',
        'route' => 'route_ex_01',
    ]
],

productor

// 给不同路由发消息，需要声明多个信道

$channel_1 = Amqp::channel('channel_1');
CLog::info('run method: '.__METHOD__);
for($i=0; $i<10; $i++)
{
    $channel_1->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_01' );
}

1-1-N bean配置（省略其他配置）

'type' => AMQPExchangeType::DIRECT,
'channels' => [
    'channel_1'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_01',
        'queue' => 'que_01',
        'route' => 'route_ex_01',
    ],
    'channel_2'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_01',
        'queue' => 'que_02',
        'route' => 'route_ex_01',
    ],
],

productor

// 给不同路由发消息，需要声明多个信道

$channel_1 = Amqp::channel('channel_1');
$channel_2 = Amqp::channel('channel_2');
CLog::info('run method: '.__METHOD__);
for($i=0; $i<10; $i++)
{
    // 这里是$channel_1还是$channel_2，都是一样的效果
    $channel_1->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_01' );
}

发送结果	que_01	que_02
数量	10	10

1-N-N bean配置（省略其他配置）

'type' => AMQPExchangeType::DIRECT,
'channels' => [
    'channel_1'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_01',
        'route' => 'route_ex_01',
        'queue' => 'que_01',
    ],
    'channel_2'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_01',
        'route' => 'route_ex_02',
        'queue' => 'que_02',
    ],
],

productor

// 给不同路由发消息，需要声明多个信道

$channel_1 = Amqp::channel('channel_1');
$channel_2 = Amqp::channel('channel_2');
CLog::info('run method: '.__METHOD__);
for($i=0; $i<10; $i++)
{
    // 这里是$channel_1还是$channel_2，都是一样的效果
    $channel_1->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_01' );
    $channel_2->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_02' );
}

下面这样也是可以获得相同的发送结果，Amqp::channel()是为了载入信道配置的

$channel_1 = Amqp::channel('channel_1');
// 如果注释掉下一行，则不会发送到que_02
$channel_1 = Amqp::channel('channel_2');
CLog::info('run method: '.__METHOD__);
for($i=0; $i<10; $i++)
{
    // 这里是$channel_1还是$channel_2，都是一样的效果
    $channel_1->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_01' );
    $channel_1->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_02' );
}

发送结果	que_01	que_02
数量	10	10

如果$channel_2在push时，填入route_ex_01，则发送结果如下

发送结果	que_01	que_02
数量	20	0

N-1-1 bean配置（省略其他配置）

'type' => AMQPExchangeType::DIRECT,
'channels' => [
    'channel_1'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_01',
        'route' => 'route_ex_01',
        'queue' => 'que_01',
    ],
    'channel_2'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_02',
        'route' => 'route_ex_01',
        'queue' => 'que_01',
    ],
],

productor

// 给不同路由发消息，需要声明多个信道

$channel_1 = Amqp::channel('channel_1');
$channel_2 = Amqp::channel('channel_2');
CLog::info('run method: '.__METHOD__);
for($i=0; $i<10; $i++)
{
    $channel_1->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_01' );
    $channel_2->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_01' );
}

|发送结果|ex_01|ex_02|que_01
|–|–|–|
|数量|0|20|20|

在交换机中查看消息，是按照发送顺序排列；但是在消费消息队列时，很大可能出现不按发送顺序读取队列消息的情况
虽然建立了两个信道，但是却被最后的信道覆盖了exchange名，消息全部从最后一个交换机通过

N-N-1 bean配置（省略其他配置）

'type' => AMQPExchangeType::DIRECT,
'channels' => [
    'channel_1'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_01',
        'route' => 'route_ex_01',
        'queue' => 'que_01',
    ],
    'channel_2'=>[  //对应的key即为channel_id
        'exchange' => 'exchange_02',
        'route' => 'route_ex_02',
        'queue' => 'que_01',
    ],
],

productor

// 给不同路由发消息，需要声明多个信道

$channel_1 = Amqp::channel('channel_1');
$channel_2 = Amqp::channel('channel_2');
CLog::info('run method: '.__METHOD__);
for($i=0; $i<10; $i++)
{
    $channel_1->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_01' );
    $channel_2->push('hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_02' );
}

发送结果	ex_01	ex_02	que_01
数量	0	20(In)/10(Out)	10

虽然建立了不同的交换机，但是从结果看，所有信息都是通过ex_02发送，因此是20(In)。在转发到队列时，因为找不到route_ex_01，所以只发送出去10(Out)
如果是通过不同的生产者就可以正常推送到同一个queue中，适合N-1-1和N-N-1，如下所示

第一种，建立不同的生产者方法

// 调用
DirectDao::product1();

public static function product1()
    {
        $channel_1 = Amqp::channel('channel_2');

        CLog::info('run method: '.__METHOD__);
        for($i=0; $i<10; $i++)
        {
            $channel_1->push('from:ex_1|hey!-----'.$i.date('Y-m-d h:i:s'),[],'route_ex_02' );
            sleep(2);
        }

        $name = 'steve';
        return context()->getResponse()->withContent('Hello' . ($name === '' ? '' : ", {$name}"));
    }

第二种，通过传入不同的信道信息、路由，构建生产者信道

DirectDao::product('channel_1', 'route_ex_01');

public static function product($channel, $route_key)
    {
        $channel_1 = Amqp::channel($channel);

        CLog::info('run method: '.__METHOD__);
        for($i=0; $i<10; $i++)
        {
            $channel_1->push('from:'.$channel.'|hey!-----'.$i.date('Y-m-d h:i:s'),[],$route_key );
            sleep(2);
        }

        $name = 'steve';
        return context()->getResponse()->withContent('Hello' . ($name === '' ? '' : ", {$name}"));
    }

关于swoft-amqp的一些简单应用

云平台相关栏目本月热门文章