czapp-api/Home/Lib/Action/CronAction.class.php

<?php



class CronAction extends Action {
			


 
	public  function createEplate_index(  ){
		$this->createElectronicPlate();
	}
	
 
	private  function createElectronicPlate(  ){
		$start = time();
		$pendingElectricPlate = Redis('jyzl_wait_create_eplate');
		$plateLocalPath = Redis('jyzl_wait_upload2oss_eplate', 'queue');
		while( (time() - $start) < 60 ){
			$licensPlate = $pendingElectricPlate->pop();
			if($licensPlate){
				echo 'pop licensPlate:'.$licensPlate.PHP_EOL;
				$field = 'LicensPlate,VehicleColor, FullName, Address, FrameNumber, MotorNumber, VehicleBrand';
				$vehicleInfo = M('jms_vehicle')->where(array('LicensPlate' => $licensPlate))->field($field);
				$localPath = '生成函数';
				$up2ossWait = json_encode(array('licensPlate' => $licensPlate, 'localPath' => $localPath));
				$plateLocalPath->push($up2ossWait);
			}
			sleep(1);
		}
	}
	
 
	private  function uploadElectronicPlate2Oss(  ){
		$config = array(
			'OssDsn' => C('OSS_DSN'),
			'filePathDir' => '/data/wwwroot/czapp.rltest.cn/1.0.0//uploadimage/',
			"SaveRule" => "/electronic_plate/{Y}{m}{d}/{uid}.{ext}",
			"AllowExts" => array('jpg', 'png', 'jpeg'), // 允许上传的文件后缀（留空为不限制）
			"ResizeImage" => true,   // 是否自动压缩
			"MaxImageWidth" => 1024,
			"MaxImageHeight" => 1024,
			"IsCheckRgb" => true,
			"MinImgAverageRgb" => 70
		);
		$upload = new \Jms\File\Oss2($config);
		
		$start = time();
		$plateLocalPath = Redis('jyzl_wait_upload2oss_eplate');
		while( (time() - $start) < 60 ){
			$data = $plateLocalPath->pop();
			var_dump($data);
			json_decode($data,true);
			var_dump($data);
			if($data){
				//TEST
				$fileName = 'gravatar.jpg';
				//$fileName = $data['localPath'];
				$uploadRes = $upload->localFileUpload($fileName);
				if(!$uploadRes['success']){
					echo $uploadRes['message'].PHP_EOL;
					continue;
				}
				$frontImageUrl = $uploadRes['objectname'];
		
				$updateRes = M('jms_vehicle')->where(array('LicensePlate' => $data['licensPlate']))->setField('FrontElectronicPlateUrl',$frontImageUrl);
				if(!$updateRes){
					echo 'update failed'.PHP_EOL;
					continue;
				}
				if(file_exists($filename)){
					unlink($filename);
				}
			}
			sleep(1);
		}
	}
	
 
	public  function uploadEplate_index(  ){
		$this->uploadElectronicPlate2Oss();
	}
	
 
	public  function test_function(  ){
		$plate = 'BJ000100';
		$pendingElectricPlate = Redis("jyzl_wait_create_eplate","queue");
		$licensePlate = $plate;
		$pendingElectricPlate->push($licensePlate);
		
	}
	
 
	public  function acrossAlarm2Kafka(  ){
		// 从 topic ：gps_location_data 取轨迹
		$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);
		// 订阅轨迹数据topic
		$consumer->subscribe(['gps_location_data']);
		while (true) {
		    $message = $consumer->consume(120*1000);
		    switch ($message->err) {
		        case RD_KAFKA_RESP_ERR_NO_ERROR:
		           // 判断是否超出围栏范围 ，存入 topic：gps_alarm_msg_queue 
						 $route_info = json_decode($message->payload,true);
						 if( empty($route_info) ){
							 echo 'empty route info.';
							 break;
						 }
						 $result = $this->produceAcrossAlarmData($route_info);
						 if($result){
							echo $result,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 produceAcrossAlarmData( $route_info ){
		if( !$route_info['DeviceId'] ){
			return 'device id is not exists.';
		}
		if( !$route_info['Longitude'] ){
			return 'longitude is not exists.';
		}
		if( !$route_info['Latitude'] ){
			return 'latitude is not exists.';
		}
		if( !$route_info['DeviceTime'] ){
			return 'device time is not exists.';
		}
		// 从数据库中取出车牌号，缓存1天
		if( S('plate-'.$route_info['DeviceId']) ){
			$plate = S('plate-'.$route_info['DeviceId']);
		}else{
			$where = array('GpsDeviceNumber'=>$route_info['DeviceId']);
			$plate = M('jms_vehicle')->where($where)->getField('LicensePlate');
			S('plate-'.$route_info['DeviceId'], $plate, 24*60*60);
		}
		// 是否启用围栏
		$rlfd_vehicle_fence = Redis('rlfd_vehicle_fence','hash');
		$fence = $rlfd_vehicle_fence->get($plate);
		$fence = json_decode($fence, true);
		
		if( empty($fence) ){
			return '围栏信息不存在';
		}
		if( !$fence['fenceStatus'] ){
			return '未启用围栏';
		}
		$fence_info = $fence['fenceInfo'];
		if( empty($fence_info['data']) ){
			return '围栏坐标数据不存在';
		}
		// 是否越界
		$route_point = array(
			'lng' => $route_info['Longitude'],
			'lat' => $route_info['Latitude']
		);
		$result = true; // 默认在围栏内
		if( $fence_info['type'] == 'circle' ){ // 圆形围栏
			$distance = \Jms\Algo\Geometry::distanceBetween2BdPoints($fence_info['data']['center'], $route_point); //获取轨迹点到围栏中心点间距离,km
			$result = $distance*1000 < $fence_info['data']['radius'];// 距离圆心大于半径说明越界了
		}elseif( $fence_info['type'] == 'polygon' ){ // 多边形围栏
			$result = \Jms\Algo\Geometry::isInPolygon($fence_info['data']['vertex'], $route_point);
		}else{
			return '未知围栏类型';
		}
		if( !$result ){
			$alarm_data = array(
			  "type" => C('FENCE_ALARM'),
			  "title" => "超出电子围栏",
			  "content" => "车辆 {$plate} 已超出设置的电子围栏范围，请前往停车处确认是否被盗。",
			  "device_number" => $route_info['DeviceId']
			);
			kafkaProducer('gps_alarm_msg_queue', $alarm_data); // 添加到kafka
			return '添加告警消息到 gps_alarm_msg_queue ';
		}
		
		return '没有超出围栏';
		
	}
	
 
	public  function mockProduce(  ){
		$msg_data = array(
			'DeviceId' => FFFFFF123122,
			'State' => 1,
			'Speed' => 1.2,
			'Longitude' => 121.20638,
			'Latitude' => 30.18852,
			'DeviceTime' => date('Y-m-d H:i:s'),
			'LBS' => 'LBS',
			'Direction' => 's',
		);
		$msg_data = json_encode($msg_data, JSON_UNESCAPED_UNICODE);
		kafkaProducer('gps_location_data',$msg_data);
	}
	

}