use App\Libraries\Stomp\Stomp;

    protected function sendToMQ($destination, $msg_data, $persistent = false)
    {
        try {
            $con = new Stomp(config('app.mq_url'));
            $con->connect();
            $con->begin("Transaction");
            $con->send($destination, json_encode($msg_data), array('persistent'=> $persistent));
            $con->commit("Transaction");
            $con->disconnect();
        } catch (\Exception $e) {
            app('log')->warn($e->getMessage());
        }
    }

        try {
            $this->consumer = new Stomp($this->activemq_uri); //$this->acitvemq_uri就是上面的activeMq地址app.mq_url
            if (!$this->consumer->isConnected()) {
                $this->consumer->connect();
                $this->consumer->setReadTimeout(3);
            }
            app('log')->info($this->log_remark . "connect to active mq success");
        } catch (StompException $e) {
            app('log')->info("connect to active mq failed : " . $e->getMessage());
            die();
        }
        $queue = self::getQueue();//得到队列名称，上面定义的destination
        $this->consumer->subscribe($queue);  //订阅
        //循环读帧
        while ($this->consumer->hasFrameToRead()) {
            try {
                $message = $this->consumer->readFrame();
                $data = json_decode($message->body, true);  // 这里其实就是得到了上面的队列消息msg_data
                //验证数据并更新数据
                $handle_result = self::removeDuplicateExpressInfo($data['traces'], $data['shipping_id']);
                if ($handle_result) {
                    $this->consumer->ack($message);
                }
            } catch (\Exception $e) {
                app('log')->info("handle with message failed : " . $e->getMessage());
                if (!$this->consumer->isConnected()) {
                    $this->consumer->connect();
                } else {
                    break;
                }
            }
        }
        $this->consumer->unsubscribe($queue);    //释放订阅
        $this->consumer->disconnect();      //端口连接