|
|
@@ -0,0 +1,190 @@
|
|
|
+<?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;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+}
|
git