修改

Home/Lib/Action/CatchFdRouteKafkaAction.class.php

@@ -296,5 +296,124 @@ class CatchFdRouteKafkaAction extends Action {
 		return array('success'=>false,'message'=>'handle wifi_macs failed');
 	}
 	
+ 
+	public  function heart_index(  ){
+		$broker_list = C('KAFKA_BROKER_LIST');
+		if (empty($broker_list)) {
+			exit("KAFKA_BROKER_LIST must be config!".PHP_EOL);
+		}
+		$group = C('WXT_DATA_KAFKA_GROUP');
+		if (empty($group)) {
+			exit("WXT_DATA_KAFKA_GROUP must be config!".PHP_EOL);
+		}
+		$topics = C('C61_HEARTBEAT_DATA_KAFKA_TOPIC');
+		if (empty($topics)) {
+			exit("C61_HEARTBEAT_DATA_KAFKA_TOPIC must be config!".PHP_EOL);
+		}
+		$topics = explode(',',$topics);
+		$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', $group);
+		// Initial list of Kafka brokers（添加 kafka集群服务器地址）
+		
+		$conf->set('metadata.broker.list', $broker_list);
+		$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', 'earliest');
+		// Set the configuration to use for subscribed/assigned topics
+		$conf->setDefaultTopicConf($topicConf);
+		$consumer = new RdKafka\KafkaConsumer($conf);
+		// 订阅轨迹数据topic
+		$consumer->subscribe($topics);
+		while (true) {
+		    $message = $consumer->consume(120*1000);
+		    switch ($message->err) {
+		        case RD_KAFKA_RESP_ERR_NO_ERROR:
+						 $data = json_decode($message->payload,true);
+						 if( $data ){
+							 //var_dump($data);
+							 $result = $this->addC61DataToTb($data);
+							 if (APP_DEBUG && isset($result['message'])) {
+								 echo $result['message'] . PHP_EOL;
+							 }
+						 }
+		            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;
+		    }
+		}
+	}
+	
+ 
+	private  function addC61DataToTb( $data ){
+		if(!is_array($data)){
+			return array('success'=>false,'message'=>'addRfidDataToTb failed,data must be array!');
+		}
+		if(!$data['tid']){
+			return array('success'=>false,'message'=>'addRfidDataToTb failed,imei not existed!');
+		}
+		$device_info = $this->getDeviceInfo($data['tid']);
+		
+		if (empty($device_info)) {;
+			return false;
+		}
+		// 存入表
+		$heartbeat_data = array(
+			'imei' => $data['tid'],
+			'rsrp' => $data['rsrp'],
+			'battery_level' => $data['battery_level'],
+			'temperature' => $data['temperature'],
+			'timestamp' => $data['timestamp'],
+			'label_2g4_id' => $data['label_2g4_id'],
+			'sw_ver' => $data['sw_ver'],
+			'device_info' => $data['device_info'],
+			'iccid' => $data['iccid'],
+		);
+		$save_data=array('battery_level'=> $data['battery_level'],'iccid'=> $data['iccid']);
+		if($device_info['online_time']>$device_info['wifi_online_time']){
+			$save_data['online_time']=$data['timestamp'];
+		}else{
+			$save_data['wifi_online_time']=$data['timestamp'];
+		}
+		M('devices')->createSave(array('imei'=>$data['tid']),$save_data);
+		
+		
+		$res = M('heartbeat_log')->createAdd($heartbeat_data);
+		//var_dump($res);
+		if(!$res['success']){
+			return array('success'=>false,'添加心跳日志失败');
+		}
+		return array('success'=>true,'message'=>'succes');
+	}
+	
 
 }