1
<?php
2

3
declare(strict_types=1);
4
/**
5
 * This file is part of Hyperf.
6
 *
7
 * @link     https://www.hyperf.io
8
 * @document https://hyperf.wiki
9
 * @contact  group@hyperf.io
10
 * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
11
 */
12
use Hyperf\Amqp\IO\IOFactory;
13

14
use function Hyperf\Support\env;
15

16
// 默认配置 如果需要项目隔离vhost 赋值新数组并更改vhost参数项
17
$default_config = [
18
    'host' => env('AMQP_HOST', 'localhost'),
19
    'port' => (int) env('AMQP_PORT', 5672),
20
    'user' => env('AMQP_USER', 'guest'),
21
    'password' => env('AMQP_PASSWORD', 'guest'),
22
    'vhost' => env('AMQP_VHOST', 'center'),
23
    'open_ssl' => false,
24
    'concurrent' => [
25
        // 单个消费者进程内的并发消费上限
26
        'limit' => 2,
27
    ],
28
    'pool' => [
29
        // 连接池数量
30
        'connections' => 2,
31
    ],
32
    'io' => IOFactory::class,
33
    'params' => [
34
        'insist' => false,
35
        'login_method' => 'AMQPLAIN',
36
        'login_response' => null,
37
        'locale' => 'en_US',
38
        'connection_timeout' => 10,
39
        // 尽量保持是 heartbeat 数值的两倍
40
        'read_write_timeout' => 60,
41
        'context' => null,
42
        // 网络层保活 与心跳互补 兜底检测
43
        'keepalive' => true,
44
        // 尽量保证每个消息的消费时间小于心跳时间
45
        'heartbeat' => 30,
46
        'channel_rpc_timeout' => 0.0,
47
        // 是否在对象析构时自动关闭连接
48
        'close_on_destruct' => false,
49
        // 连接内可复用的空闲 Channel 数上限
50
        'max_idle_channels' => 10,
51
        'connection_name' => null,
52
    ],
53
];
54

55
// 如果需要隔离项目vhost
56
// 商家
57
$merchant_config = $default_config;
58
$merchant_config['vhost'] = env('MERCHANT_AMQP_VHOST', 'merchant');
59

60
return [
61
    'enable' => true,
62
    'default' => $default_config,
63
    'merchant' => $merchant_config,
64
];

1
# RabbitMQ
2
AMQP_HOST=127.0.0.1
3
AMQP_PORT=5672
4
AMQP_USER=guest
5
AMQP_PASSWORD=guest
6
AMQP_VHOST=center
7
MERCHANT_AMQP_VHOST=merchant

1
<?php
2
declare (strict_types=1);
3

4
namespace App\Common\Enum;
5
class AmqpEnum
6
{
7
    // AMQP重试次数
8
    const AMQP_RETRY_COUNT = 3;
9

10
    const STATUS_INIT = 0;
11
    const STATUS_PADDING = 1;
12
    const STATUS_SUCCESS = 2;
13
    const STATUS_FAILED = 3;
14
    const STATUS_DESC = [
15
        self::STATUS_INIT => '待处理',
16
        self::STATUS_PADDING => '处理中',
17
        self::STATUS_SUCCESS => '处理成功',
18
        self::STATUS_FAILED => '处理失败'
19
    ];
20

21
    // 默认vhost配置项
22
    const DEFAULT_VHOST_CONFIG = 'default';
23

24
    // 商家vhost配置项
25
    const MERCHANT_VHOST_CONFIG = 'merchant';
26

27
    // 测试队列配置名称
28
    const DEMO_QUEUE_NAME = 'demo';
29
    // 商家端 测试队列配置名称
30
    const M_DEMO_QUEUE_NAME = 'm_demo';
31

32

33

34
}

1
<?php
2
declare(strict_types=1);
3

4
use App\Amqp\Producer\Merchant\DyCallBackProducer;
5
use Hyperf\Amqp\Message\Type;
6
use function Hyperf\Support\env;
7
use App\Amqp\Producer\Demo\DemoProducer;
8
use App\Common\Enum\AmqpEnum;
9

10
$env = env('APP_ENV', 'dev');
11
return [
12
    // 默认配置
13
    AmqpEnum::DEFAULT_VHOST_CONFIG => [
14
        'vhost' => env('AMQP_VHOST', 'center'),
15
        'queues' => [
16
            // 命名规则 队列名称.项目名称（默认center）.环境
17
            AmqpEnum::DEMO_QUEUE_NAME => [
18
                'exchange' => "test.demo.center.{$env}",
19
                'exchange_type' => Type::DIRECT->value, // 交换机类型 direct fanout topic headers
20
                'queue' => "test.demo.queue.center.{$env}", // 队列名
21
                'routing_key' => "test.demo.key.center.{$env}", // 路由键
22
                'producer' => DemoProducer::class, // 生产者
23
            ]
24
        ]
25
    ],
26
    // 商家配置
27
    AmqpEnum::MERCHANT_VHOST_CONFIG => [
28
        'vhost' => env('MERCHANT_AMQP_VHOST', 'merchant'),
29
        'queues' => [
30
            // 命名规则 队列名称.项目名称（merchant）.环境
31
            AmqpEnum::M_DEMO_QUEUE_NAME => [
32
                'exchange' => "test.m_demo.merchant.{$env}",
33
                'exchange_type' => Type::FANOUT->value, // 交换机类型 direct fanout topic headers
34
                'queue' => "test.m_demo.queue.merchant.{$env}", // 队列名
35
                'routing_key' => "test.m_demo.key.merchant.{$env}", // 路由键
36
                'producer' => DemoProducer::class, // 生产者
37
            ]
38
        ]
39
    ],
40
];

1
<?php
2
declare (strict_types=1);
3

4
namespace App\Common\Enum;
5

6
class CacheEnum
7
{
8
    // 防止队列重复消费键
9
    const CHECK_QUEUE_REPEAT_CONSUMING_KEY = 'check_queue_repeat_consuming';
10
}

1
CREATE TABLE `ufo_m_amqp_dead_letter_message` (
2
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
3
  `message_id` varchar(255) NOT NULL DEFAULT '' COMMENT '消息ID',
4
  `exchange` varchar(255) NOT NULL DEFAULT '' COMMENT '原始交换机',
5
  `routing_key` varchar(255) NOT NULL DEFAULT '' COMMENT '原始路由键',
6
  `queue_name` varchar(255) NOT NULL DEFAULT '' COMMENT '队列名称',
7
  `queue_config` varchar(255) NOT NULL DEFAULT '' COMMENT '队列配置名（区分业务）',
8
  `exchange_type` varchar(255) NOT NULL DEFAULT '' COMMENT '交换机类型：direct fanout topic headers',
9
  `message_content` json  COMMENT '消息内容',
10
  `retry_count` int(11) unsigned NOT NULL DEFAULT '0' COMMENT '重试次数',
11
  `last_retry_time` datetime DEFAULT NULL COMMENT '最后重试时间',
12
  `status` tinyint(4) unsigned NOT NULL DEFAULT '0' COMMENT '状态(0-待处理,1-处理中,2-处理成功,3-处理失败)',
13
  `error_msg` json DEFAULT NULL COMMENT '错误信息',
14
  `created_at` datetime DEFAULT NULL COMMENT '创建时间',
15
  `updated_at` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
16
  `deleted_at` datetime DEFAULT NULL,
17
  PRIMARY KEY (`id`),
18
  KEY `idxes_queue_config_status` (`queue_config`,`status`),
19
  KEY `idx_create_time` (`created_at`)
20
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Amqp失败队列消息表';

1
<?php
2
declare (strict_types=1);
3

4
namespace App\Common\Trait;
5

6
use App\Common\Enum\AmqpEnum;
7
use App\Common\Enum\CacheEnum;
8
use App\Module\Merchant\Repository\AmqpDeadLetterMessageRepository as MerchantAmqpDeadLetterMessageRepository;
9
use Hyperf\Amqp\Constants;
10
use Hyperf\Amqp\Result;
11
use Hyperf\Di\Annotation\Inject;
12
use Hyperf\Redis\Redis;
13
use Hyperf\Snowflake\IdGeneratorInterface;
14
use PhpAmqpLib\Message\AMQPMessage;
15
use PhpAmqpLib\Wire\AMQPTable;
16
use function Hyperf\Config\config;
17

18
trait AmqpMessageTrait
19
{
20
    #[Inject]
21
    public Redis $redis;
22

23
    #[Inject]
24
    public MerchantAmqpDeadLetterMessageRepository $merchantAmqpDeadLetterMessageRepository;
25

26
    #[Inject]
27
    public IdGeneratorInterface $idGenerator;
28

29
    /**
30
     * 生成队列配置
31
     * @param string $pool_name 队列池名称
32
     * @param string $queue_name 队列配置名称
33
     * @param string $message_id 唯一消息ID
34
     * @return array
35
     */
36
    public function getConfig(string $pool_name, string $queue_name, string $message_id = ''): array
37
    {
38
        $poolName = $pool_name;
39

40
        // 队列配置
41
        $queue_config = config("amqp_queue.{$pool_name}.queues." . $queue_name);
42
        $exchange = $queue_config['exchange'];
43
        $type = $queue_config['exchange_type'];
44
        $routingKey = $queue_config['routing_key'];
45
        $queue = $queue_config['queue'];
46
        $properties = [
47
            'content_type' => 'text/plain',
48
            'delivery_mode' => Constants::DELIVERY_MODE_PERSISTENT,
49
            'timestamp' => time(),
50
            'application_headers'=> new AMQPTable([
51
                'retry' => '0', // 重试次数
52
                'message_id' => strval($message_id ?: $this->idGenerator->generate()), // 消息唯一ID
53
                'queue_config' => $queue_name,
54
                'exchange_type' => (string) $type,
55
                'queue_name' => $queue_config['queue'],
56
                'vhost_config' => $pool_name
57
            ]),
58
        ];
59

60
        return compact('poolName', 'exchange', 'type', 'routingKey', 'properties', 'queue');
61
    }
62

63

64
    /**
65
     * 检查消息是否重复消费
66
     * @param AMQPMessage $msg
67
     * @return bool
68
     * @throws \RedisException
69
     */
70
    public function checkMessageRepeat(AMQPMessage $msg): bool
71
    {
72
        $msgData = $this->messageHandle($msg, '', false);
73
        $message_id = $msgData['message_id'];
74
        $vhost_config = $msgData['vhost_config'];
75
        // 重试的话不检查重复消费
76
        if (empty($message_id) || ($msgData['retry_count'] ?? 0) > 0) {
77
            return true;
78
        }
79
        $redis_key = CacheEnum::CHECK_QUEUE_REPEAT_CONSUMING_KEY . ":{$vhost_config}:" . config('app_env') . ":{$message_id}";
80

81
        // 一天内不能重复消费
82
        return $this->redis->setnx($redis_key, $message_id) && $this->redis->expire($redis_key, 24 * 3600);
83
    }
84

85

86
    /**
87
     * 获取队列消息详细信息
88
     * @param AMQPMessage $msg
89
     * @param string $error_msg
90
     * @param bool $needGenerateId 不存在message_id时是否生成
91
     * @return array
92
     */
93
    public function messageHandle(AMQPMessage $msg, string $error_msg, bool $needGenerateId = true): array
94
    {
95
        $exchange = $msg->getExchange();
96
        $routing_key = $msg->getRoutingKey();
97
        $message_content = $msg->getBody();
98
        //headersObject 是一个AMQPTable对象
99
        $headersObject = $msg->get_properties()['application_headers'];
100
        //调用getNativeData()得到一个数组
101
        $headersArray  = $headersObject->getNativeData();
102

103
        if ($needGenerateId && !isset($headersArray['message_id'])) {
104
            $headersArray['retry'] = 0;
105
            $headersArray['message_id'] = strval($this->idGenerator->generate());
106
        }
107

108
        $message_id = $headersArray['message_id'];
109
        $queue_name = $headersArray['queue_name'] ?? '';
110
        $queue_config = $headersArray['queue_config'] ?? '';
111
        $exchange_type = $headersArray['exchange_type'] ?? '';
112
        $vhost_config = $headersArray['vhost_config'] ?? '';
113
        $retry_count = ($headersArray['retry'] ?? 0);
114
        $last_retry_time = date('Y-m-d H:i:s');
115
        $error_msg = compact('error_msg');
116

117
        return compact('exchange', 'routing_key', 'message_id', 'queue_name', 'error_msg', 'vhost_config',
118
            'queue_config', 'exchange_type', 'retry_count', 'headersArray', 'message_content', 'last_retry_time');
119
    }
120

121
    /**
122
     * 丢弃消息
123
     * @param AMQPMessage $msg
124
     * @param string $error_msg 错误信息
125
     * @return Result
126
     */
127
    public function rejectHandle(AMQPMessage $msg, string $error_msg): Result
128
    {
129
        $msgData = $this->messageHandle($msg, $error_msg);
130
        $msgData['status'] = AmqpEnum::STATUS_FAILED;
131

132
        // 记录失败日志
133
        $this->saveDlxMessage($msgData);
134
        return Result::DROP;
135
    }
136

137
    /**
138
     * 确认应答
139
     * @param AMQPMessage $msg
140
     * @return Result
141
     */
142
    public function ackHandle(AMQPMessage $msg): Result
143
    {
144
        $msgData = $this->messageHandle($msg, '');
145
        $msgData['status'] = AmqpEnum::STATUS_SUCCESS;
146

147
        // 更新失败日志
148
        $this->saveDlxMessage($msgData, false);
149
        return Result::ACK;
150
    }
151

152
    /**
153
     * 手动重试
154
     * @param AMQPMessage $msg
155
     * @param string $error_msg 错误信息
156
     * @param int $retry_num 重试次数
157
     * @return Result
158
     */
159
    public function retryHandle(AMQPMessage $msg, string $error_msg, int $retry_num = AmqpEnum::AMQP_RETRY_COUNT): Result
160
    {
161
        $msgData = $this->messageHandle($msg, $error_msg);
162
        $exchange      = $msgData['exchange'];
163
        $routing_key   = $msgData['routing_key'];
164
        $channel       = $msg->getChannel();
165
        $body          = $msgData['message_content'];
166
        $headersArray  = $msgData['headersArray'];
167

168

169
        if ($headersArray['retry'] < $retry_num) {
170
            $headersArray['retry']++;
171
            // 重新入队列
172
            $channel->basic_publish(
173
                new AMQPMessage($body, [
174
                    'content_type' => 'text/plain',
175
                    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
176
                    'timestamp' => time(),
177
                    'application_headers'=> new AMQPTable($headersArray)
178
                ]),
179
                $exchange,
180
                $routing_key
181
            );
182
            // 因为重新入队了 所以这条消息直接确认
183
            $result = Result::ACK;
184
            $msgData['status'] = AmqpEnum::STATUS_PADDING;
185
        } else {
186
            // 丢弃
187
            $result = Result::DROP;
188
            $msgData['status'] = AmqpEnum::STATUS_FAILED;
189
        }
190
        // 记录失败日志
191
        $this->saveDlxMessage($msgData);
192
        return $result;
193
    }
194

195
    /**
196
     * 保存失败队列消息
197
     * @param array $saveData
198
     * @param bool $needInsert 记录不存在是否新增
199
     * @return bool
200
     */
201
    public function saveDlxMessage(array $saveData, bool $needInsert = true): bool
202
    {
203
        if (empty($saveData) || empty($saveData['message_id'])) {
204
            return false;
205
        }
206
        $saveWhere = [
207
            'message_id' => $saveData['message_id'],
208
            'queue_config' => $saveData['queue_config'],
209
        ];
210

211
        $saveData['update_time'] = date('Y-m-d H:i:s');
212
        $saveData['error_msg'] = $saveData['error_msg'] ?? [];
213

214

215
        $saveData['message_content'] = empty($saveData['message_content']) ? null : (is_string($saveData['message_content'])
216
            ? json_decode($saveData['message_content'], true) : $saveData['message_content']);
217

218
        // 根据不同项目 写入不同表
219
        $vhost_config = $saveData['vhost_config'] ?? AmqpEnum::DEFAULT_VHOST_CONFIG;
220
        $repository = $this->getDlxMessageRepository($vhost_config);
221

222
        return $repository->insertOrUpdate($saveWhere, $saveData, $needInsert);
223
    }
224

225

226
    /**
227
     * 根据不同vhost配置 返回不同项目的失败消息表
228
     * @param string $vhost_config
229
     */
230
    public function getDlxMessageRepository(string $vhost_config)
231
    {
232
        switch ($vhost_config) {
233
            case AmqpEnum::MERCHANT_VHOST_CONFIG:
234
                // 商家端
235
                $repository = $this->merchantAmqpDeadLetterMessageRepository;
236
                break;
237
            default:
238
                $repository = $this->merchantAmqpDeadLetterMessageRepository;
239
        }
240
        return $repository;
241
    }
242
}

1
/**
2
 * 插入或更新
3
 * @param array $where
4
 * @param array $data
5
 * @param bool $needInsert 记录不存在是否新增
6
 * @return bool
7
 */
8
public function insertOrUpdate(array $where, array $data, bool $needInsert = true): bool
9
{
10
    // 如果传入了 $where 条件，尝试查找记录
11
    if (!empty($where)) {
12
        $record = AmqpDeadLetterMessage::query()->where($where)->first();
13
        if ($record) {
14
            // 记录存在，执行更新
15
            return $record->fill($data)->save();
16
        }
17
    }
18

19
    // 记录不存在，执行新增
20
    if ($needInsert) {
21
        return (new AmqpDeadLetterMessage())->fill($data)->save();
22
    }
23
    return true;
24
}

1
protected array $fillable = ['id', 'message_id', 'exchange', 'routing_key', 'queue_name', 'queue_config', 'exchange_type', 'message_content', 'retry_count', 'last_retry_time', 'status', 'error_msg', 'created_at', 'updated_at', 'deleted_at'];
2

3

4
protected array $casts = ['id' => 'integer', 'retry_count' => 'integer', 'status' => 'integer', 'created_at' => 'datetime', 'updated_at' => 'datetime', 'message_content' => 'array', 'error_msg' => 'array'];

1
<?php
2

3
declare(strict_types=1);
4

5
namespace App\Amqp\Producer\Demo;
6

7
use App\Common\Enum\AmqpEnum;
8
use App\Common\Trait\AmqpMessageTrait;
9
use Hyperf\Amqp\Message\ProducerMessage;
10

11
/**
12
 * 测试生产者
13
 * Class DemoProducer
14
 * @package App\Amqp\Producer\Demo
15
 */
16
class DemoProducer extends ProducerMessage
17
{
18
    use AmqpMessageTrait;
19
    public function __construct($data, string $message_id = '')
20
    {
21
        $this->setConfig($message_id);
22
        // 需要发送的数据
23
        $this->payload = $data;
24
    }
25

26
    /**
27
     * 设置队列配置
28
     * @param string $message_id
29
     * @return void
30
     */
31
    public function setConfig(string $message_id = ''): void
32
    {
33
        $params = $this->getConfig(AmqpEnum::DEFAULT_VHOST_CONFIG, AmqpEnum::DEMO_QUEUE_NAME, $message_id);
34
        fillProperty($params, $this);
35
    }
36
}

1
if (!function_exists('fillProperty')) {
2
    /**
3
     * 使用反射填充属性
4
     * @param array $params
5
     * @param object $instance
6
     */
7
    function fillProperty(array $params, object $instance): void
8
    {
9
        $ref = new ReflectionObject($instance);
10
        foreach ($params as $key => $value) {
11
            if ($ref->hasProperty($key)) {
12
                $prop = $ref->getProperty($key);
13
                $prop->setAccessible(true); // 允许访问 protected/private
14
                $prop->setValue($instance, $value);
15
            }
16
        }
17
    }
18
}

1
<?php
2
declare (strict_types=1);
3

4
namespace App\Controller;
5

6
use App\Amqp\Producer\Demo\DemoProducer;
7
use Hyperf\Amqp\Producer;
8
use Hyperf\Di\Annotation\Inject;
9
use Hyperf\HttpServer\Annotation\Controller;
10
use Hyperf\HttpServer\Annotation\PostMapping;
11

12
#[Controller(prefix: 'demo')]
13
class DemoController extends AbstractController
14
{
15
    #[Inject]
16
    protected Producer $producer;
17

18
    /**
19
     * demo 生产者
20
     * @return array
21
     */
22
    #[PostMapping(path: 'demoProduce')]
23
    public function demoProduce()
24
    {
25
        $res = $this->producer->produce(new DemoProducer(['tenant_id' => 1, 'product_list' => [1,2]], 'test_message_id'));
26

27
        return [
28
            'message' => $res ? "发送成功" : '发送失败',
29
        ];
30
    }
31
}

1
<?php
2

3
declare(strict_types=1);
4

5
namespace App\Amqp\Consumer\Demo;
6

7
use App\Amqp\Consumer\BasicConsumer;
8
use App\Common\Enum\AmqpEnum;
9
use App\Exception\BusinessException;
10
use Hyperf\Amqp\Annotation\Consumer;
11
use Hyperf\Amqp\Result;
12
use PhpAmqpLib\Exception\AMQPChannelClosedException;
13
use PhpAmqpLib\Exception\AMQPConnectionBlockedException;
14
use PhpAmqpLib\Exception\AMQPConnectionClosedException;
15
use PhpAmqpLib\Exception\AMQPIOException;
16
use PhpAmqpLib\Message\AMQPMessage;
17

18
#[Consumer(name: "DemoConsumer", nums: 1)]
19
class DemoConsumer extends BasicConsumer
20
{
21
    public function __construct()
22
    {
23
        $this->setConfig();
24
    }
25

26
    /**
27
     * 设置队列配置
28
     * @return void
29
     */
30
    public function setConfig(): void
31
    {
32
        $params = $this->getConfig(AmqpEnum::DEFAULT_VHOST_CONFIG, AmqpEnum::DEMO_QUEUE_NAME);
33
        fillProperty($params, $this);
34
    }
35

36
    public function consumeMessage($data, AMQPMessage $message): Result
37
    {
38
        $json_data = json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
39
        try {
40

41
            $this->logger->info('接收到消息：' . $json_data);
42

43
            // 检查消息是否重复消费
44
            if (!$this->checkMessageRepeat($message)) {
45
                throw new BusinessException("消息重复消费,本次跳过");
46
            }
47

48
            // todo 业务处理
49

50

51
            // 确认应答
52
            return $this->ackHandle($message);
53
        }  catch (BusinessException|AMQPIOException|AMQPConnectionClosedException|AMQPConnectionBlockedException|AMQPChannelClosedException|\Exception $e) {
54
            if ($e instanceof BusinessException) {
55
                $this->logger->error("[测试消费者]发生业务性异常 消息丢弃：" . $json_data . "，错误原因：" . var_export($e->getMessage(), true));
56
                $error_msg = "[业务异常]错误原因：" . var_export($e->getMessage(), true);
57
                // 丢弃消息
58
                return $this->rejectHandle($message, $error_msg);
59
            } else {
60
                $this->logger->error("[测试消费者]发生致命异常 消息重新入队：" . $json_data . "，错误原因：" . var_export($e->getMessage(), true));
61
                $error_msg = "[致命异常]错误原因：" . var_export($e->getMessage(), true);
62
                // 手动重试机制
63
                return $this->retryHandle($message, $error_msg);
64
            }
65
        }
66

67
    }
68

69
}

1
<?php
2
namespace App\Exception;
3

4
use Throwable;
5

6
class BusinessException extends \RuntimeException
7
{
8
    protected $code = 400;
9

10
    public function __construct($message = "", $code = 0, Throwable $previous = null)
11
    {
12
        $this->code = $code ?: $this->code;
13
        parent::__construct($message, $this->code, $previous);
14
    }
15
}

1
<?php
2
declare (strict_types=1);
3

4
namespace App\Amqp\Consumer;
5

6
use App\Common\Trait\AmqpMessageTrait;
7
use Hyperf\Amqp\Message\ConsumerMessage;
8
use Hyperf\Di\Annotation\Inject;
9
use Psr\Log\LoggerInterface;
10

11
class BasicConsumer extends ConsumerMessage
12
{
13
    use AmqpMessageTrait;
14

15
    #[Inject]
16
    protected LoggerInterface $logger;
17

18
    /**
19
     * 判断是否允许自启动
20
     * @return bool
21
     */
22
    public function isEnable(): bool
23
    {
24
        return true;
25
    }
26
}

1
<?php
2
declare (strict_types=1);
3

4
namespace App\Command;
5

6
use App\Command\Service\DeadLetterMessageService;
7
use App\Common\Enum\AmqpEnum;
8
use Hyperf\Command\Annotation\Command;
9
use Hyperf\Command\Command as HyperfCommand;
10
use Hyperf\Di\Annotation\Inject;
11
use Psr\Log\LoggerInterface;
12
use Symfony\Component\Console\Input\InputArgument;
13
use Symfony\Component\Console\Input\InputOption;
14

15
/**
16
 * 失败消息重新入队
17
 * Class DeadLetterMessageCommand
18
 * @package App\Command
19
 */
20
#[Command(
21
    name: "dlxMessage:requeue",
22
    arguments:[
23
        ['vhost_config', InputArgument::REQUIRED, 'vhost配置名称：default|merchant'],
24
        ['message_ids', InputArgument::REQUIRED, '消息ID，多个用英文逗号隔开']
25
    ],
26
    options:[
27
        ['debug', null, InputOption::VALUE_OPTIONAL, '是否调试模式', '0']
28
    ],
29
    description:"失败消息重新入队")
30
]
31
class DeadLetterMessageCommand extends HyperfCommand
32
{
33
    #[Inject]
34
    protected LoggerInterface $logger;
35

36
    #[Inject]
37
    protected DeadLetterMessageService $service;
38

39
    public function handle()
40
    {
41
        try {
42
            $start = microtime(true);
43
            $vhost_config = $this->input->getArgument('vhost_config') ?: AmqpEnum::DEFAULT_VHOST_CONFIG;
44
            $message_ids = $this->input->getArgument('message_ids');
45
            $debug = $this->input->getOption('debug');
46

47
            $message_ids = !empty($message_ids) ? explode(',', $message_ids) : [];
48

49
            $this->service->requeueDlxMessages($vhost_config, $message_ids, $debug == 1);
50

51
            $duration = round(microtime(true) - $start, 2);
52
            $this->info("失败消息重新入队操作完成，耗时：{$duration}秒");
53
        } catch (\Throwable $e) {
54
            // 简单记录错误
55
            $this->logger->error('失败消息重新入队系统错误', [
56
                'error' => $e->getMessage(),
57
                'file' => $e->getFile(),
58
                'line' => $e->getLine()
59
            ]);
60
            $this->info('失败消息重新入队系统错误！' . var_export($e->getMessage(), true));
61
        }
62

63
    }
64
}

1
<?php
2

3
namespace App\Command\Service;
4

5
use App\Common\Enum\AmqpEnum;
6
use App\Common\Enum\CacheEnum;
7
use App\Common\Trait\AmqpMessageTrait;
8
use App\Exception\BusinessException;
9
use Hyperf\Amqp\Producer;
10
use Hyperf\Context\ApplicationContext;
11
use Hyperf\Di\Annotation\Inject;
12
use Hyperf\Redis\Redis;
13
use Psr\Log\LoggerInterface;
14
use function Hyperf\Config\config;
15

16
class DeadLetterMessageService
17
{
18
    use AmqpMessageTrait;
19

20
    #[Inject]
21
    protected LoggerInterface $logger;
22

23
    #[Inject]
24
    protected Producer $producer;
25

26
    #[Inject]
27
    public Redis $redis;
28

29
    /**
30
     * 重新入队死信消息
31
     * @param string $vhost_config vhost配置名称
32
     * @param array $message_ids 消息ID
33
     * @param bool $debug 是否调试模式
34
     * @return void
35
     */
36
    public function requeueDlxMessages(string $vhost_config, array $message_ids, bool $debug): void
37
    {
38
        if (empty($message_ids) || empty($vhost_config)) {
39
            $this->logger->error('参数非法');
40
            throw new BusinessException('参数非法');
41
        }
42

43
        $repository = $this->getDlxMessageRepository($vhost_config);
44
        $list = $repository->getListByMessageIds($message_ids);
45
        $container = ApplicationContext::getContainer();
46

47
        if (empty($list)) {
48
            $this->logger->error('没有符合条件的死信消息');
49
            throw new BusinessException('没有符合条件的死信消息');
50
        }
51

52
        foreach ($list as $item) {
53
            try {
54
                $item_message_id = strval($item['message_id']);
55
                if (!$debug) {
56
                    // 删除防止重复消费键
57

58
                    $redis_key = CacheEnum::CHECK_QUEUE_REPEAT_CONSUMING_KEY . ":{$vhost_config}:" . config('app_env') . ":{$item_message_id}";
59
                    $this->redis->del($redis_key);
60

61

62
                    // 获取队列配置
63
                    $queue_config = config("amqp_queue.{$vhost_config}.queues." . $item['queue_config']);
64

65
                    $producer = $container->make($queue_config['producer'], [
66
                        $item['message_content'] ?: [],
67
                        $item_message_id
68
                    ]);
69

70
                    // 重新入队列
71
                    $this->producer->produce($producer);
72
                }
73

74
                // 更新死信消息为处理中 因为重新入队了
75
                $saveData = [
76
                    'message_id' => $item_message_id,
77
                    'queue_config' => $item['queue_config'],
78
                    'retry_count' => 0,
79
                    'last_retry_time' => date('Y-m-d H:i:s'),
80
                    'status' => AmqpEnum::STATUS_PADDING,
81
                    'error_msg' => ['error_msg' => '']
82
                ];
83
                !$debug && $this->saveDlxMessage($saveData, false);
84

85
            } catch (\Exception $e) {
86
                $this->logger->error("[死信消息重回队列异常]消息ID:{$item['message_id']},队列配置为：{$item['queue_config']}" .
87
                    "消息内容为：" . json_encode($item['message_content'], JSON_UNESCAPED_UNICODE) .
88
                    "，错误原因:" . var_export($e->getMessage(), true));
89
            }
90
        }
91
    }
92
}

1
/**
2
 * 根据消息ID获取列表
3
 * @param array $message_ids
4
 * @return array
5
 */
6
public function getListByMessageIds(array $message_ids): array
7
{
8
    $list = AmqpDeadLetterMessage::query()->whereIn('message_id', $message_ids)->get();
9
    if ($list->isEmpty()) {
10
        return [];
11
    }
12
    return $list->toArray();
13
}