RabbitMQ的实现

点击查看RabbitMQ教程

生产者：

$config = [
    'host' => '127.0.0.1',
    'vhost' => 'rabbitmqhost',
    'port' => 5672,
    'login' => 'admin',
    'password' => '123456'
];

$exchangeName = 'kd_sms_send_ex'; //交换机名
$queueName = 'kd_sms_send_q'; //队列名称
$routingKey = 'sms_send'; //路由关键字(也可以省略)

// 建立生产者与mq之间的连接
$conn = new AMQPStreamConnection(
    $config['host'], $config['port'], $config['login'], $config['password'], $config['vhost']
);
// 在已连接基础上建立生产者与mq之间的通道
$ch = $conn->channel();

// 声明初始化交换机
$ch->exchange_declare($exchangeName, 'direct', false, true, false);
// 声明初始化一条队列
$ch->queue_declare($queueName, false, true, false, false);
// 将队列与某个交换机进行绑定，并使用路由关键字
$ch->queue_bind($queueName, $exchangeName, $routingKey);

$msgBody = json_encode(["name" => "iGoo", "age" => 22]);
// 生成消息
$msg = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]);  
// 推送消息到某个交换机
$ch->basic_publish($msg, $exchangeName, $routingKey);

消费者：

$config = [
    'host' => '127.0.0.1',
    'vhost' => 'rabbitmqhost',
    'port' => 5672,
    'login' => 'admin',
    'password' => '123456'
];

// 交换机名
$exchangeName = 'kd_sms_send_ex'; 
// 队列名称
$queueName = 'kd_sms_send_q'; 
// 路由关键字(也可以省略)
$routingKey = 'sms_send'; 

// 建立生产者与mq之间的连接
$conn = new AMQPStreamConnection(
    $config['host'], $config['port'], $config['login'], $config['password'], $config['vhost']
);
// 在已连接基础上建立生产者与mq之间的通道
$ch = $conn->channel();

// 声明初始化一条队列
$ch->queue_declare($queueName, false, true, false, false);

// 在接收消息的时候调用回调函数
$ch->basic_consume($queueName, '', false, true, false, false, function ($msg){
   vdump(" [x] Received ", $msg->body, "\n");
});

// 堵塞
while(count($ch->callbacks)) {
   $ch->wait();
}