nbfd_tp3/Home/Lib/Action/NingboFangdaoSyncDataAction.class.php

<?php



class NingboFangdaoSyncDataAction extends Action {
			


 
	public  function kafka_index(  ){
		
		$broker_list = C('KAFKA_BROKER_LIST');
		
		if (empty($broker_list)) {
			exit("KAFKA_BROKER_LIST must be config!".PHP_EOL);
		}
		$group = C('SYNC_DATA_KAFKA_GROUP');
		if (empty($group)) {
			exit("SYNC_DATA_KAFKA_GROUP must be config!".PHP_EOL);
		}
		$topics = C('SYNC_DATA_KAFKA_TOPIC');
		if (empty($topics)) {
			exit("SYNC_DATA_KAFKA_TOPIC must be config!".PHP_EOL);
		}
		$topics = explode(',',$topics);
		
		// 从 topic ：rlstation_rfid_location 取轨迹
		$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', 'latest');
		// Set the configuration to use for subscribed/assigned topics
		$conf->setDefaultTopicConf($topicConf);
		$consumer = new RdKafka\KafkaConsumer($conf);
		
		
		
		// 订阅轨迹数据topic
		$consumer->subscribe($topics);
		
		$config = C('ORACLE_CONFIG');
		if (empty($config)) {
			exit("ORACLE_CONFIG must be config!".PHP_EOL);
		}
		$host= $config['host'];
		$port=  $config['port'];
		$instance_name= $config['instance_name'];
		$username= $config['username'];
		$password=  $config['password'];
		
		/*
		$host= '192.168.100.23';
		$port=   '1521';
		$instance_name= 'helowin';
		$username=  'DSSC3';
		$password=  'Rliandssc3';
		*/
		$conn = oci_connect($username, $password, $host.':'.$port.'/'. $instance_name,'AL32UTF8');
		if (!$conn) {
		   $e = oci_error();
		   trigger_error(htmlentities($e['message'], ENT_QUOTES), E_USER_ERROR);
		}
		
		while (true) {
			//var_dump($conn);
		    $message = $consumer->consume(120*1000);
		
		    switch ($message->err) {
		        case RD_KAFKA_RESP_ERR_NO_ERROR:
						
						 $data = json_decode($message->payload,true);
						 if( $data ){
							
							$res=$this->addDataToNingbo($data,$conn);
							if(!$res['success']){
								throw new \Exception($res['message']);
							}
		
							 //$this->addRfidDataToRenlian($data);
						 }
		            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:
					    echo "default break";
						$this->debug_log( 'default_Log', $message->errstr() );
						$this->debug_log( 'default_Log', $message->err );
		            break;
		    }
		}
	}
	
 
	private  function addDataToNingbo( $data, $conn ){
		$type_arr=explode('_',$data['DATA_TYPE']);
		if($type_arr[0]=='vehicle'){
			 switch ($type_arr['1']) {
		      case 'save':
				   $sql='declare
		            tagId number;
		        	begin
		            INSERT INTO DSSC3.W_DW_RFID_TAGS("ID", "RFID_SN","INSTALLER","INSTA_DATE") VALUES (DSSC3.SEQ_W_DW_RFID_TAGS.nextval, \''.$data['RFID_SN'].'\',\''.$data['INSTALLER'].'\', TO_DATE(\''.$data['INSTA_DATE'].'\', \'SYYYY-MM-DD HH24:MI:SS\'))  returning ID into tagId;
		            INSERT INTO DSSC3.W_DW_NON_MOTOR("ID", "RFID_ID","PLATE_NO","CAR_BRAND","CAR_TYPE") VALUES (DSSC3.SEQ_W_DW_RFID_TAGS.nextval, tagId,\''.$data['PLATE_NO'].'\',\''.$data['CAR_BRAND'].'\',\''.$data['CAR_TYPE'].'\');
		        	end;';
		        $stid = oci_parse($conn, $sql);
		        $r = oci_execute($stid);
		        oci_free_statement($stid);
		        break;        
		      case 'update':
					//更新
		        $sql='UPDATE DSSC3.W_DW_NON_MOTOR o,DSSC3.W_DW_RFID_TAGS s SET s.RFID_SN = \''.$params['RFID_SN'].'\' WHERE s.PLATE_NO =\''.$params['OLD_NO'].'\'  AND o.rfid_id = s.id  ';
		        $stid = oci_parse($conn, $sql);
		        $r = oci_execute($stid);
		        $sql='UPDATE DSSC3.W_DW_NON_MOTOR  SET PLATE_NO = \''.$params['PLATE_NO'].'\',CAR_BRAND=\''.$params['CAR_BRAND'].'\',CAR_TYPE=\''.$params['CAR_TYPE'].'\' WHERE PLATE_NO = \''.$params['OLD_NO'].'\' ';
		        $stid2 = oci_parse($conn, $sql);
		        $r2 = oci_execute($stid2);
		      	  break;
				case 'delete':
				 	$rfid_ids=[];
		 		   $sql='SELECT RFID_ID FROM DSSC3.W_DW_NON_MOTOR  WHERE PLATE_NO in ('.$data['PLATE_NO'].')';
		     	  $stid = oci_parse($conn, $sql);
		     	  $r = oci_execute($stid);
		  		 	while($row = oci_fetch_array($stid, OCI_ASSOC+OCI_RETURN_NULLS)) {
		            $rfid_ids[]=$row['RFID_ID'];
		  			}
				 	$sql='DELETE FROM DSSC3.W_DW_RFID_TAGS WHERE ID in ('.$rfid_ids.')';
				 	$stid = oci_parse($conn, $sql);
				 	$r = oci_execute($stid);
				 	$sql='DELETE FROM DSSC3.W_DW_NON_MOTOR WHERE PLATE_NO in ('.$data['PLATE_NO'].')';
				 	$stid = oci_parse($conn, $sql);
				 	$r = oci_execute($stid);
				 	break;
		      default:
		     	  break;
		   }
		}elseif($type_arr[0]=='station'){
			 switch ($type_arr['1']) {
		      case 'save':
				   M('stations')->createAdd($data);
		        break;        
		      case 'update':
					$where=array(
						'id'=>$data['update_id'],
					);
				 	M('stations')->createSave($where,$data);
		      	  break;
				case 'delete':
				 	//18610,18609
				 	$deleteIds=explode(',',$date['delete_id']);
				 	M('stations')->where(['id'=>['in',$$deleteIds]])->delete();
				 	break;
		      default:
		     	  break;
		   }
		}
	}
	

}