push() ->setPlatform('all') ->addAllAudience() ->setNotificationAlert('车辆异动告警'); try { $response = $push_payload->send(); print_r($response); } catch (\JPush\Exceptions\APIConnectionException $e) { // try something here print $e; } catch (\JPush\Exceptions\APIRequestException $e) { // try something here print $e; } } public function pushFromKafka( ){ $conf = new RdKafka\Conf(); // Set a rebalance callback to log partition assignments (optional)(当有新的消费进程加入或者退出消费组时,kafka 会自动重新分配分区给消费者进程,这里注册了一个回调函数,当分区被重新分配时触发) $conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) { switch ($err) { case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS: echo "Assign: "; var_dump($partitions); $kafka->assign($partitions); break; case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: echo "Revoke: "; var_dump($partitions); $kafka->assign(NULL); break; default: throw new \Exception($err); } }); // Configure the group.id. All consumer with the same group.id will consume( 配置groud.id 具有相同 group.id 的consumer 将会处理不同分区的消息,所以同一个组内的消费者数量如果订阅了一个topic, 那么消费者进程的数量多于这个topic 分区的数量是没有意义的。) // different partitions. $conf->set('group.id', 'myConsumerGroup'); // Initial list of Kafka brokers(添加 kafka集群服务器地址) $conf->set('metadata.broker.list', '127.0.0.1'); $topicConf = new RdKafka\TopicConf(); // Set where to start consuming messages when there is no initial offset in // offset store or the desired offset is out of range. // 'smallest': start from the beginning $topicConf->set('auto.offset.reset', 'smallest'); // Set the configuration to use for subscribed/assigned topics $conf->setDefaultTopicConf($topicConf); $consumer = new RdKafka\KafkaConsumer($conf); // 消费者订阅 $consumer->subscribe(['gps_alarm_msg_queue']); while (true) { $message = $consumer->consume(120*1000); switch ($message->err) { case RD_KAFKA_RESP_ERR_NO_ERROR: $msg_data = json_decode($message->payload,true); if($msg_data){ // 使用极光推送消息 $jpush_client = new \JPush\Client( C('JPUSH_APP_KEY'), C('JPUSH_MASTER_SECRET') ); // 电子围栏告警,上锁车辆异动告警,低电量告警,被盗告警,广播消息推送 $this->jpushMsg( $jpush_client, $message->payload ); } break; case RD_KAFKA_RESP_ERR__PARTITION_EOF: echo "No more messages; will wait for more\n"; break; case RD_KAFKA_RESP_ERR__TIMED_OUT: echo "Timed out\n"; break; default: throw new \Exception($message->errstr(), $message->err); break; } } } public function jpushMsg( $client, $msg_data ){ /* $msg_data = '{ "type":2, "title":"被盗告警", "content":"车被偷了,赶紧去找", "device_number":"FFFFFF123122" }'; $msg_data = '{ "type":9, "title":"群推消息", "content":"这是一个广播" }'; */ $single_push_type = array( C('LOCK_VEHICLE_ALARM'), // 1-锁车告警 C('STOLEN_ALARM'), // 2-被盗告警 C('FENCE_ALARM'), // 3-电子围栏告警 C('LOWWER_BATTERY_ALARM'), // 4-低电量告警 ); $msg_data = json_decode($msg_data,true); if(!$msg_data){ echo 'invalid message data format!'.$msg_data . PHP_EOL; return; } // 通过传过来的车牌查出JgClientRegistrationId if( empty($client) ){ $client = new \JPush\Client( C('JPUSH_APP_KEY'), C('JPUSH_MASTER_SECRET') ); } if( in_array($msg_data['type'],$single_push_type) && $msg_data['device_number'] ){ $where = array('DeviceNumber|GpsDeviceNumber' => $msg_data['device_number']); $fields= 'CityId,LicensePlate,FullName,JgClientRegistrationId'; if( !S( 'jpush_vinfo_'.$msg_data['device_number'] ) ){ //如果不存在,则缓存 $vehicle_info = M('jms_vehicle')->field($fields)->where($where)->find(); S( 'jpush_vinfo_'.$msg_data['device_number'], $vehicle_info, 60 ); //缓存1分钟 } $vehicle_info = S( 'jpush_vinfo_'.$msg_data['device_number'] ); //从缓存取 } // 推送平台 $platform = array('ios', 'android'); // 通知栏显示内容 $alert = $msg_data['content']; // 推送android $android_notification = array( 'title' => $msg_data['title'], 'builder_id' => 2, //通知栏样式 ); // 推送ios $ios_notification = array( 'sound' => 'default', 'badge' => '+1', 'content-available' => true ); // 可选项 $options = array( 'sendno' => 100, 'time_to_live' => 100, ); // 组装推送 $cid = $client->push()->getCid(); //var_dump($cid['body']['cidlist'][0]); $push_payload = $client->push() ->setCid($cid['body']['cidlist'][0]) ->setPlatform( $platform ) ->iosNotification($alert, $ios_notification) ->androidNotification($alert, $android_notification) ->options($options); if( $msg_data['type'] == C('BROADCASTING') ){ // 9-广播 $push_payload->addAllAudience(); }elseif( isset($vehicle_info) && $vehicle_info ){ // 单推 $msg_data['vehicle_info'] = $vehicle_info; $push_payload->addRegistrationId($vehicle_info['JgClientRegistrationId']); }else{ //未知类型 var_dump('出错了:'.$msg_data); return; } try { $response = $push_payload->send(); // 保存日志 $msg_data['response'] = array( 'code' => $response['http_code'], 'resp_msg' => 'ok', ); $this->saveLog($msg_data); print_r($response); } catch (\JPush\Exceptions\APIConnectionException $e) { // try something here $msg_data['response'] = array( 'code' => $e->getHttpCode(), 'resp_msg' => substr($e->__toString(), strpos($e->__toString(),'[')), ); $this->saveLog($msg_data); print $e; } catch (\JPush\Exceptions\APIRequestException $e) { // try something here $msg_data['response'] = array( 'code' => $e->getHttpCode(), 'resp_msg' => substr($e->__toString(), strpos($e->__toString(),'[')), ); $this->saveLog($msg_data); print $e; } } public function saveLog( $msg_data ){ $fdls_app_message = M('jms_baojing_message'); $log_data = array( 'ID' => create_guid(), 'Type' => $msg_data['type'], 'Title' => $msg_data['title'], 'Comment' => $msg_data['content'], 'SendStatus' => $msg_data['response']['code'], 'AddTime' => date('Y-m-d H:i:s'), 'RespMsg' => $msg_data['response']['resp_msg'], ); if( $msg_data['type'] != C('BROADCASTING') ){ // 非广播 $log_data['CityId'] = $msg_data['vehicle_info']['CityId']; $log_data['DeviceNumber'] = $msg_data['device_number']; $log_data['FullName'] = $msg_data['vehicle_info']['FullName']; $log_data['LicensePlate'] = $msg_data['vehicle_info']['LicensePlate']; } $result = $fdls_app_message->createAdd($log_data); return $result; } public function kafkaProducer( $msg_data ){ if (!extension_loaded('rdkafka')){ echo 'rdkafka extension is not installed!!'.PHP_EOL; return false; } /********************* 初始化生产者配置项 start **************************/ //考勤记录分析结果生产者 $rk = new RdKafka\Producer(); $rk->setLogLevel(LOG_DEBUG); $rk->addBrokers("127.0.0.1"); $start = microtime(true); $topic = $rk->newTopic('gps_alarm_msg_queue'); /********************* 初始化生产者配置项 end **************************/ if( empty($msg_data) ){ var_dump($msg_data); return; } if( is_array($msg_data) || is_object($msg_data) ){ // 对象、数组转换 $msg_data = json_encode($msg_data); } $topic->produce(RD_KAFKA_PARTITION_UA, 0, $msg_data); $rk->poll(0); } public function mockProduce( ){ $msg_data = '{ "type":9, "title":"群推消息", "content":"这是一个广播" }'; $this->kafkaProducer($msg_data); } }