use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage;
config.php 文件
"order_dispose" => [ "routingKey" => "xxx", "name" => "xxx", //"class" => \app\rmq\pull\xxx::class, "exchange" => [ "name" => "xxx", "type" => "topic", #exchange类型,可以是direct、topic、headers或fanout之一。 "passive" => false, #如果设置为true,则不会创建exchange,仅检查是否存在指定名称的exchange。 "durable" => true, #如果设置为true,则exchange将在重启后继续存在。 "auto_delete" => false, #如果设置为true,则exchange将在最后一个绑定被删除后自动删除。 "internal" => false,#如果设置为true,则exchange将用于内部用途,不允许客户端发布消息到该exchange。 ], /*"queue" => [ "name" => "xxx", #队列名称 "passive" => false, #如果为true,则不会创建队列。如果队列不存在,则会返回一个错误 "durable" => true, #如果为true,则队列将被持久化。当RabbitMQ服务器重启时,队列将保留下来 "exclusive" => false, #如果为true,则队列只能被当前连接使用,当连接关闭时队列将被删除。该选项通常用于创建临时队列。 "auto_delete" => false, #如果设置为true,则exchange将在最后一个绑定被删除后自动删除。 ],*/
//实现方式 /** * @throws \Exception */ public static function publisherMessage($data) { // json格式数据 $msgBody = json_encode($data); // 日志记录 Log::info("rabbitmq推送订单信息" . $msgBody); try { $config = config('rabbitmq.AMQP'); // 创建连接和通道 $connection = new AMQPStreamConnection($config["host"], $config["port"], $config["user"], $config["password"], $config["vhost"], false, 'AMQPLAIN', null, 'en_US', 30, 60, null, false, 30); $channel = $connection->channel(); // 声明交换机 $listen_config = config("rabbitmq.Listening.order_dispose"); $channel->exchange_declare($listen_config["exchange"]["name"], $listen_config["exchange"]["type"], $listen_config["exchange"]["passive"], $listen_config["exchange"]["durable"], $listen_config["exchange"]["auto_delete"], $listen_config["exchange"]["internal"]); // 创建消息 $properties = ['content_type' => 'application/json', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]; $msg = new AMQPMessage($msgBody, $properties); // 发送消息到交换机 $channel->basic_publish($msg, $listen_config["exchange"]["name"], $listen_config["routingKey"]); //echo " [x] Sent '$msgBody'\n"; // 关闭通道和连接 $channel->close(); $connection->close(); } catch (\Exception $e) { Log::info("MQ推送订单数据失败,订单信息:" . $msgBody . ',错误信息:' . $e->getMessage()); } }