ActiveMq获取topic数据

<?php// error_reporting(0);error_reporting(E_ERROR | E_WARNING | E_PARSE);set_time_limit(0);$broker = 'tcp://localhost:61613';$queue  = '/topic/test';$stomp = new Stomp($broker);  //php需安装stomp扩展，引入stomp类$stomp->subscribe($queue);while(true) {//判断是否有读取的信息if($stomp->hasFrame()) {$frame = $stomp->readFrame();$data = json_decode($frame->body, true);//保存原始接收日志file_put_contents("test.txt",json_encode($data)."\n",FILE_APPEND);  $mysqli = new mysqli("localhost", "root","root","test");  //地址，数据库账号，数据库密码，数据库名称$val=$data;$val['create_time']=time();$bkgPicUrl=explode('/pic',$val['bkgPicUrl']);$facePicUrl=explode('/pic',$val['facePicUrl']);$val['bkgPicUrl']='http://18.0.0.149:6501/pic'.$bkgPicUrl[1];$val['facePicUrl']='http://18.0.0.149:6501/pic'.$facePicUrl[1];$click_video="that.$parent.changeData('showControlToast', true,1);";  //关联监控$val['humans']=json_encode($val['humans']);$coordinate_1=$row['longitude'].','.$row['latitude'];$query="insert into activemq(alarm_id,index_code,camera_name,age,sex,glass,smile,ethnic,bkg_pic_url,face_pic_url,similarity,status,cascade_alarm_id,alarm_time,control_id,humans,human_id,human_name,list_lib_id,coordinate,create_time) values('{$val["alarmId"]}','{$val["indexCode"]}','{$val["cameraName"]}','{$val["age"]}','{$val["sex"]}','{$val["glass"]}','{$val["smile"]}','{$val["ethnic"]}','{$val["bkgPicUrl"]}','{$val["facePicUrl"]}','{$val["similarity"]}','{$val["status"]}','{$val["cascadeAlarmId"]}','{$val["alarmTime"]}','{$val["controlId"]}','{$val["humans"]}','{$val["humanId"]}','{$val["humanName"]}','{$val["listLibId"]}',\"{$coordinate_1}\",'{$val["create_time"]}');";$mysqli->query($query);$mysqli->close();//储存日志输出echo "\n\ttime:".date("Y-m-d H:i:s",$val['create_time'])."\talarm_id:".$val['alarmId'];//表示消息被处理掉了，ack()函数很重要$stomp->ack($frame);//exit();}sleep(1);}