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());}
}