admin_consumer/app/Amqp/Consumer/MqConsumer.php

<?php

declare (strict_types = 1);

namespace App\Amqp\Consumer;

use App\JsonRpc\ChatServiceInterface;
use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Amqp\Result;
use Hyperf\Di\Annotation\Inject;
use PhpAmqpLib\Message\AMQPMessage;
use Psr\Log\LoggerInterface;
use Hyperf\Redis\RedisFactory;

#[Consumer(exchange: 'hyperf', routingKey: 'hyperf', queue: 'hyperf', name: "MqConsumer", nums: 1)]
class MqConsumer extends ConsumerMessage
{
    /**
     * @var ChatServiceInterface
     */
    #[Inject]
    private $chatServiceClient;

    protected $logger;
    #[Inject]
    protected RedisFactory $redisFactory;

    public function __construct(LoggerInterface $logger)
    {
        $this->logger = $logger;
    }

    public function consumeMessage($data, AMQPMessage $message): Result
    {
        $redis = $this->redisFactory->get('default');
        $lockKey = 'mq_consumer_lock';
        $lockValue = uniqid();

        // 获取锁
        $lockAcquired = $redis->setnx($lockKey, $lockValue);
        if (!$lockAcquired) {
            $this->logger->info('Another consumer instance is already running.');
            var_dump($redis->get($lockKey));
            return Result::ACK;
        }

        // 设置锁过期时间，防止死锁
        $redis->expire($lockKey, 60);
        try {
            // 数据存储
            $this->logger->info('消费数据', ['data' => $data]);
            // 调用数据处理服务
            $result = $this->chatServiceClient->addChatRecords($data);
            // 记录处理结果
            $this->logger->info("消费成功：", ['result' => $result]);
            return Result::ACK;
        } catch (\Exception $e) {
            // 记录错误
            $this->logger->error("消费失败：", ['error' => $e->getMessage()]);
            // 返回拒绝，重新入队
            // return Result::REQUEUE;
        }

    }

    public function isEnable(): bool
    {
        return true;
    }
}