消费等@lbk@2018/02/07

api/config/main.php

@@ -35,14 +35,15 @@ return [
                 ],
             ],
         ],
-        /*'request' => [
-            'class' => '\yii\web\Request',
-            'enableCookieValidation' => false,
-            'parsers' => [
-                'application/json' => 'yii\web\JsonParser',
-            ],
+        'request' => [
+            //'class' => '\yii\web\Request',
+            //'enableCookieValidation' => false,
+            //'parsers' => [
+            //    'application/json' => 'yii\web\JsonParser',
+            //],
+			'baseUrl' => '/api',
             'cookieValidationKey' => 'O1d232trde1xww-M97_7QvwPo-5QGdkLMp#@#@',
-        ],*/
+        ],
         /*'errorHandler' => [
             'errorAction' => 'site/error',
         ],*/

backend/config/main.php

@@ -14,6 +14,7 @@ return [
     'modules' => [],
     'components' => [
         'request' => [
+			'baseUrl' => '',
             'csrfParam' => '_csrf-backend',
         ],
         'user' => [
@@ -36,7 +37,7 @@ return [
         ],
 		'errorHandler' => [
 			'class' => 'components\ErrorHandler',
-			'errorAction' => '/error/error',
+			'errorAction' => 'site/error',
 		],
         'db' =>   require __DIR__ . '/../../common/config/db.php',
 
@@ -45,8 +46,8 @@ return [
             'enableStrictParsing' => false,  //不启用严格解析
             'showScriptName' => false,   //隐藏index.php
             'rules' => [
-                '<module:\w+>/<controller:\w+>/<id:\d+>' => '<module>/<controller>/view',
-                '<controller:\w+>/<id:\d+>' => '<controller>/view',
+                /*'<module:\w+>/<controller:\w+>/<id:\d+>' => '<module>/<controller>/view',
+                '<controller:\w+>/<id:\d+>' => '<controller>/view',*/
             ],
         ]
     ],

backend/controllers/MessageController.php

@@ -1,12 +1,11 @@
 <?php
 namespace backend\controllers;
 
-use backend\models\Jobs;
 use backend\forms\MessageForm;
+use common\helpers\KeyHelper;
 use common\logic\Amqp\Cache;
-use common\logic\Amqp\Message;
-use common\logic\Amqp\Queue;
 use components\Exception;
+use components\service\AmqpConfig;
 use Yii;
 
 class MessageController extends BaseController
@@ -14,17 +13,23 @@ class MessageController extends BaseController
 	/**
 	 * [查询回执信息]
 	 * @author: libingke
+	 * @param $query
+	 * @param $queue
+	 * @return bool
+	 * @throws Exception
 	 */
-	public function actionQueryReceipt($query)
+	public function actionQueryReceipt($query, $queue)
 	{
-		if (is_string($query)) {
-			$status = Cache::getData($query);
+		if (is_string($query) && is_string($query)) {
+			$statusKey = KeyHelper::getMessageStatusKey($query, $queue);
+			$resultKey = KeyHelper::getMessageResultKey($query, $queue);
+			$status = Yii::$app->redis->get($statusKey);
 
 			if ($status !== false) {
-				$mark = Cache::getMarkById($status);//状态码 => 提示信息
+				$mark = AmqpConfig::getMarkById($status);//状态码 => 提示信息
 
 				if (Cache::STATUS_HAND_OK == $status) {
-					return Cache::getData('result_' . $query);//直接返回数据
+					return json_decode(Yii::$app->redis->get($resultKey), true);//直接返回数据
 
 				} else {
 					throw new Exception(2101, $mark);//非正常时回执
@@ -47,28 +52,13 @@ class MessageController extends BaseController
 	 */
 	public function actionSend()
 	{
-		$sign = Yii::$app->request->get('sign');
-		if ($sign == '')
-			throw new Exception(2201);
-
-		$job = Jobs::fetchDataBySign($sign, false);
-		if (!isset($job['scenario']))
-			throw new Exception(2202);//验证场景
-		$scenario = $job['scenario'];
-		$queue = $job['queue'];
-
-		if (!Yii::$app->request->isPost)
-			throw new Exception('1001');
-
 		$model = new MessageForm();
-		$model->setScenario($scenario);
-		$model->setQueue($queue);
-		$model->load(['MessageForm' => Yii::$app->request->post()]);
+		$model->setScenario('send');
+		$model->load(['MessageForm' => Yii::$app->request->get()]);
 
 		$data = [];
 		if ($model->validate()) {
-			$data['queueName'] = $model->getQueue();
-			$data['requestId'] = $model->sendMessage();
+			$data = $model->sendMessageV1_1();
 		} else {
 			$model->handleError();//处理验证失败
 		}
@@ -81,35 +71,21 @@ class MessageController extends BaseController
 	}
 
 	/**
-	 * 发送消息
+	 * 批量发送消息
 	 * @author: libingke
+	 * @return array
 	 */
 	public function actionBatchSend()
 	{
-		$sign = Yii::$app->request->get('sign');
-		if ($sign == '')
-			throw new Exception(2201);
-
-		$job = Jobs::fetchDataBySign($sign, false);
-		if (!isset($job['scenario']))
-			throw new Exception(2202);//验证场景
-		$scenario = $job['scenario'];
-		$queue = $job['queue'];
-
-		if (!Yii::$app->request->isPost)
-			throw new Exception('1001');
-
 		$model = new MessageForm();
-		$model->setScenario($scenario);
-		$model->setQueue($queue);
-		$model->load(['MessageForm' => Yii::$app->request->post()]);
+		$model->setScenario('batch_send');
+		$model->load(['MessageForm' => Yii::$app->request->get()]);
 
 		$data = [];
 		if ($model->validate()) {
-			$data['queueName'] = $model->getQueue();
-			$data['requestId'] = $model->sendMessage();
+			$data = $model->batchSendMessage();
 		} else {
-			$model->handleError();//处理验证失败
+			$model->handleError();
 		}
 
 		return [
@@ -117,126 +93,54 @@ class MessageController extends BaseController
 			'message' => Yii::t('error', 200),
 			'data' => $data
 		];
+	}
 
+	/**
+	 * 获取队列消息列表
+	 * @author: libingke
+	 */
+	public function actionList()
+	{
+		$model = new MessageForm();
+		$model->setScenario('message_list');
+		$model->load(['MessageForm' => Yii::$app->request->get()]);
 
-		$params = Yii::$app->request->post();
-		//test data start todo delete
-		$test[] = [
-			'queue' => 'y1',
-			'infos' => [
-				json_encode(['type' => 'post', 'url' => 'http://zpapi.licai.cn', 'message' => 'y1-m1']),
-				json_encode(['type' => 'post', 'url' => 'http://zpapi.licai.cn', 'message' => 'y1-m2'])
-			],
-		];
-		$test[] = [
-			'queue' => 'y2',
-			'infos' => [
-				json_encode(['type' => 'post', 'url' => 'http://zpapi.licai.cn', 'message' => 'y2-m1']),
-				json_encode(['type' => 'post', 'url' => 'http://zpapi.licai.cn', 'message' => 'y2-m2'])
-			],
-
-		];
-		$params['data'] = json_encode($test);
-		//test data end todo delete
-
-		if (!isset($params['data']))
-			throw new Exception('2005');
-
-		$data = json_decode($params['data'], true);
-		if (!is_array($data) || !$data)
-			throw new Exception('2006');
-
-		$errCount = 0;
-		foreach ($data as $items) {
-			if (!isset($items['queue'])
-				|| !is_string($items['queue'])
-				|| !isset($items['infos'])
-				|| !is_array($items['infos'])
-				|| !$items['infos']	)
-			{
-				$errCount ++;
-				continue;
-			}
-
-			try {
-				//todo check
-				//if queue exist
-				$queueModel = (new Queue())->create($items['queue']);
-				if ($queueModel['status'] == 1) {
-					$message = new Message($items['queue']);
-					foreach ($items['infos'] as $info) {
-						$message->batchSend($info, $items['queue']);
-					}
-				} else {
-					$errCount ++;
-				}
-
-			} catch (\common\logic\Amqp\Exception $e) {
-				$errCount ++;
-				//throw new Exception($e->getCode(), $e->getMessage());
-			}
+		$data = [];
+		if ($model->validate()) {
+			$data = $model->getMessageList();
+		} else {
+			$model->handleError();
 		}
 
 		return [
 			'code' => 200,
 			'message' => Yii::t('error', 200),
-			'data' => ['error_count' => $errCount]
+			'data' => $data
 		];
 	}
 
 	/**
-	 * 消费消息  todo
+	 * 消费消息
 	 * @author: libingke
 	 */
 	public function actionConsume()
 	{
-		$queue = 'login';
-		try {
-			$message = new Message($queue);
-			$message->consume($queue);
-			return $result = ['code' => 200, 'message' => Yii::t('common', 'OK')];
+		$model = new MessageForm();
+		$model->setScenario('consume');
+		$model->load(['MessageForm' => Yii::$app->request->get()]);
 
-		} catch (\common\logic\Amqp\Exception $e) {
-			$result = ['code' => $e->getCode(), 'message' => $e->getMessage()];
+		$data = [];
+		if ($model->validate()) {
+			$data = $model->consumeMessage();
+		} else {
+			$model->handleError();
 		}
 
-		return $result;
-
-	}
-
-	/**
-	 * 批量消费消息
-	 * @author: libingke
-	 */
-	public function actionBatchReceive()
-	{
-		//获取接收参数  利用model验证或者判断
-		$params = Yii::$app->request->post();
-		$post = [
-			'message1' => 'message 1',
-			'message2' => 'message 2',
-			'message3' => 'message 3',
-			'queue' => 'y1',
+		return [
+			'code' => 200,
+			'message' => Yii::t('error', 200),
+			'data' => $data
 		];
-
-		try {
-			$queue = (new Queue())->create($post['queue']);
-			if ($queue['status'] == 1) {
-				$message = new Message($queue['result']);
-				$message->receive($post['queue']);
-
-				return $result = ['code' => 200, 'message' => Yii::t('common', 'OK')];
-
-			} else {
-
-				return ['code' => $queue['status'], 'message' => $queue['result']];
-			}
-
-		} catch (\common\logic\Amqp\Exception $e) {
-			$result = ['code' => $e->getCode(), 'message' => $e->getMessage()];
-		}
-
-		return $result;
 	}
 
 	/**
@@ -245,42 +149,30 @@ class MessageController extends BaseController
 	 */
     public function actionDelete()
     {
-		//获取接收参数  利用model验证或者判断
-		$params = Yii::$app->request->post();
-		$post = [
-			'message1' => 'message 1',
-			'message2' => 'message 2',
-			'message3' => 'message 3',
-			'queue' => 'y1',
-		];
-
-		try {
-			$queue = (new Queue())->create($post['queue']);
-			if ($queue['status'] == 1) {
-				$message = new Message($queue['result']);
-				$message->delete($post['message1']);
-
-				return $result = ['code' => 200, 'message' => Yii::t('common', 'OK')];
-
-			} else {
-
-				return ['code' => $queue['status'], 'message' => $queue['result']];
-			}
-
-		} catch (\common\logic\Amqp\Exception $e) {
-			$result = ['code' => $e->getCode(), 'message' => $e->getMessage()];
-		}
-
-		return $result;
 
     }
 
 	/**
-	 * 批量删除消息
+	 * 清空消息
 	 * @author: libingke
 	 */
-	public function actionBatchDelete()
+	public function actionPurge()
 	{
+		$model = new MessageForm();
+		$model->setScenario('purge');
+		$model->load(['MessageForm' => Yii::$app->request->get()]);
 
+		$data = [];
+		if ($model->validate()) {
+			$data = $model->purge();
+		} else {
+			$model->handleError();
+		}
+
+		return [
+			'code' => 200,
+			'message' => Yii::t('error', 200),
+			'data' => $data
+		];
 	}
 }

backend/controllers/QueueController.php

@@ -0,0 +1,83 @@
+<?php
+namespace backend\controllers;
+
+use backend\forms\QueueForm;
+use components\Exception;
+use Yii;
+
+class QueueController extends BaseController
+{
+	/**
+	 * [创建队列]
+	 * @author: libingke
+	 * @return array
+	 * @throws Exception
+	 */
+	public function actionCreate()
+	{
+		if (!Yii::$app->request->isPost)
+			throw new Exception('1001');
+
+		$model = new QueueForm();
+		$model->setScenario('create_queue');
+		$model->load(['QueueForm' => Yii::$app->request->post()]);
+
+		$data = [];
+		if ($model->validate()) {
+			$data = $model->createQueue();
+		} else {
+			$model->handleError();//处理验证失败
+		}
+
+		return [
+			'code' => 200,
+			'message' => Yii::t('error', 200),
+			'data' => $data
+		];
+	}
+
+	/**
+	 * 删除队列
+	 * @author: libingke
+	 * @return array
+	 * @throws Exception
+	 */
+	public function actionDelete()
+	{
+		if (!Yii::$app->request->isPost)
+			throw new Exception('1001');
+
+		$model = new QueueForm();
+		$model->setScenario('delete_queue');
+		$model->load(['QueueForm' => Yii::$app->request->post()]);
+
+		$data = [];
+		if ($model->validate()) {
+			$data = $model->deleteQueue();
+		} else {
+			$model->handleError();
+		}
+
+		return [
+			'code' => 200,
+			'message' => Yii::t('error', 200),
+			'data' => $data
+		];
+	}
+
+
+	/**
+	 * 获取队列列表
+	 * @author: libingke
+	 */
+	public function actionList()
+	{
+		$model = new QueueForm();
+		$data = $model->getQueueList();
+		return [
+			'code' => 200,
+			'message' => Yii::t('error', 200),
+			'data' => $data
+		];
+	}
+}

backend/forms/BaseForm.php

@@ -13,6 +13,11 @@ class BaseForm extends Model
 	public function handleError()
 	{
 		$errors = $this->getFirstErrors();
-		throw new Exception(1003, reset($errors));
+		$result = reset($errors);
+		if (is_numeric($result)) {
+			throw new Exception($result, \Yii::t('error', $result));
+		} else {
+			throw new Exception(1003, $result);
+		}
 	}
 }

backend/forms/MessageForm.php

@@ -2,52 +2,168 @@
 
 namespace backend\forms;
 
+use components\service\AmqpConfig;
+use PhpAmqpLib\Exception\AMQPProtocolChannelException;
+use common\helpers\KeyHelper;
 use common\logic\Amqp\Cache;
-use common\logic\Amqp\Message;
+use components\Curl;
 use components\Exception;
+use PhpAmqpLib\Message\AMQPMessage;
+use yii\helpers\ArrayHelper;
+use Yii;
 
 class MessageForm extends BaseForm
 {
-	public $email;
+	/* send */
+	public $queue;
 
-	public $password;
+	public $message;
 
+
+	/* message_list */
+	public $name;
+
+	/**
+	 * @var integer 数量
+	 */
+	public $count;
+
+	/**
+	 * @var bool 是否重新排序
+	 */
+	public $requeue;
+
+	/**
+	 * @var string 编码
+	 */
+	public $encoding;
+
+	/**
+	 * @var string 消费id
+	 */
+	public $mid;
+
+	/**
+	 * @var string 消费类型
+	 */
 	public $type;
 
-	public $send_arr;
+	/**
+	 * @var bool 是否空消费
+	 */
+	public $do_nothing = false;
+
+	/**
+	 * @var bool 自动应答
+	 */
+	public $ack = false;
+
 
-	private $_queue = '';
+	/**
+	 * @var bool
+	 */
+	private $_stop = false;
+
+
+	/**
+	 * @var string
+	 */
+	private $_mid;
+
+	/**
+	 * @var string
+	 */
+	private $_body;
+
+	/**
+	 * @var AMQPMessage
+	 */
+	private $_message;
+
+	/**
+	 * @var array
+	 */
+	private $_rows;
 
 	/**
-	 * 登录方式
+	 * @var array
 	 */
-	const TYPE_SMS	 = 'sms';
+	private $_result;
 
-	const TYPE_EMAIL = 'email';
+	const TYPE_MID	 = 'mid';//消费某条
+	const TYPE_COUNT = 'count';//(从第一条开始) 消费条数
+	const TYPE_MC	 = 'mid_count';//(从某条开始) 消费条数
 
 	public function rules()
 	{
 		return [
-			//login
-			//[['email', 'password', 'type'], 'safe', 'on' => 'login'],
-			[['email', 'password', 'type'], 'trim', 'on' => 'login'],
-			[['email', 'password'], 'required', 'on' => 'login'],
-			['email', 'email', 'on' => 'login'],
-			['type', 'default', 'value' => static::TYPE_EMAIL, 'on' => 'login'],
-			['type', 'in', 'range' => [static::TYPE_EMAIL, static::TYPE_SMS], 'on' => 'login'],
-			//logins
-			[['send_arr'], 'required', 'on' => 'logins']
+			/* 发送消息 */
+			[['queue'], 'trim', 'on' => ['send', 'batch_send']],
+			[['queue', 'message'], 'required', 'on' => ['send', 'batch_send']],
+			['message', 'validateMessage', 'on' => ['send', 'batch_send']],
+
+			/* 获取消息列表 */
+			[['name'], 'trim', 'on' => ['message_list', 'purge', 'consume']],
+			[['name'], 'required', 'on' => ['message_list', 'purge']],
+			['count', 'integer', 'min' => 1, 'max' => 1000, 'message' => 2012,
+				'tooSmall' => 2013, 'tooBig' => 2014, 'on' => ['message_list']],
+			['count', 'default', 'value' => 20, 'on' => ['message_list']],
+			['requeue', 'boolean', 'message' => 2010, 'on' => ['message_list']],
+			['requeue', 'default', 'value' => true, 'on' => ['message_list']],
+			['encoding', 'in', 'range' => ['auto', 'base64'], 'message' => 2011, 'on' => ['message_list']],
+			['encoding', 'default', 'value' => 'auto', 'on' => ['message_list']],
+
+			/* 消费 */
+			[['type', 'name'], 'required', 'on' => ['consume']],
+			['type', 'in', 'range' => [static::TYPE_MID, static::TYPE_COUNT, static::TYPE_MC], 'on' => 'consume'],
+			['type', 'validateType', 'on' => 'consume'],
+			['mid', 'string', 'on' => 'consume'],
+			['count', 'integer', 'min' => 1, 'max' => 65000, 'message' => 2012,
+				'tooSmall' => 2013, 'on' => ['consume']],
+			['do_nothing', 'default', 'value' => false, 'on' => 'consume'],
+			['do_nothing', 'boolean',  'on' => 'consume'],
+			['ack', 'default', 'value' => false, 'on' => 'consume'],
+			['ack', 'boolean',  'on' => 'consume'],
 		];
 	}
 
-	public function setQueue($scenario)
+	public function validateMessage($attribute)
 	{
-		$this->_queue = $scenario;
+		if (!$this->$attribute)
+			throw new Exception(2001);
+
+		if (!is_array($this->$attribute))
+			throw new Exception(2002, "{$attribute} 必须是数组");
+	}
+
+	public function validateType($attribute)
+	{
+		if ($this->$attribute == static::TYPE_MC) {
+			if ($this->mid == null)
+				throw new Exception(2016);
+			if ($this->count == null)
+				throw new Exception(2017);
+
+		} else {
+			if ($this->{$this->$attribute} == null)
+				throw new Exception(2016, $this->$attribute . " 不能为空");
+		}
 	}
 
-	public function getQueue()
+	/**
+	 * [连接AMQP]
+	 * @author: libingke
+	 * @return \PhpAmqpLib\Connection\AMQPStreamConnection
+	 */
+	protected function getConnect()
 	{
-		return $this->_queue;
+		return new \PhpAmqpLib\Connection\AMQPStreamConnection(
+			Yii::$app->Amqp->host,
+			Yii::$app->Amqp->port,
+			Yii::$app->Amqp->user,
+			Yii::$app->Amqp->pass,
+			Yii::$app->Amqp->vhost
+		);
 	}
 
 	/**
@@ -55,29 +171,462 @@ class MessageForm extends BaseForm
 	 * @author: libingke
 	 * @return array
 	 * @throws Exception
+	 * @version 1.0
 	 */
 	public function sendMessage()
 	{
-		$body = json_encode([
-			'email'		=> $this->email,
-			'password'	=> $this->password,
-			'type'		=> $this->type
-		]);
-		$queue = $this->getQueue();
+		$body = call_user_func_array([$this, '_messageBody'], [$this->attributes]);
+		$mid = KeyHelper::getUniqueId('message_send');
+		$properties = [
+			'content_type' => 'text/plain',
+			'message_id' => $mid,
+			'correlation_id' => $mid,
+			'consumer_tag' => $mid
+		];
 
 		$data = [];
+		$e_name = 'message.default';
+		$k_route = 'route.default';
+		$q_name = $this->queue;
+
+		$connect  = Yii::$app->Amqp->AMQPConnection();
+		$channel  = new \AMQPChannel($connect);
+		$exchange = new \AMQPExchange($channel);
+		$exchange->setName($e_name);
+		$exchange->setType(AMQP_EX_TYPE_DIRECT);
+		$exchange->setFlags(AMQP_DURABLE);//持久化
+		$exchange->declareExchange();//声明交换机
+
+		$queue = new \AMQPQueue($channel);
+		$queue->setName($q_name);
+		$queue->declareQueue(); //声明队列
+		$queue->bind($e_name, $k_route);
+		$r = $exchange->publish($body, $k_route, AMQP_NOPARAM, $properties);
+		if ($r == true) {
+			$data['message_total'] = $queue->declareQueue();
+			$data['queue_name'] = $this->queue;
+			$data['message_new'] = [
+				'mid' => $mid,
+				'body' => $body
+			];
+
+			if ($data['message_total'] > 1) {
+				$data['message_first'] = [
+					'mid' => $queue->get()->getMessageId(),
+					'body' => $queue->get()->getBody()
+				];
+			}
+
+			Cache::setData(
+				'queue:mid:'.$mid,
+				Cache::STATUS_SEND_OK
+			);
+		} else {
+			throw new Exception(2103);
+		}
+
+		return $data;
+	}
+
+	/**
+	 * [发送消息]
+	 * @author: libingke
+	 * @return array
+	 * @throws Exception
+	 * @version 1.1
+	 */
+	public function sendMessageV1_1()
+	{
 		try {
-			$message = new Message($queue);
-			$corrId = $message->send($body, $queue);
+			//connect
+			$connect = $this->getConnect();
+			$channel = $connect->channel();
+			$this->_handleMessage($this->message);
+			//预声明
+			$channel->queue_declare($this->queue,
+				false, true, false, false);
+			$channel->basic_publish($this->_message, '', $this->queue);
+			//获取返回结果
+			list($q_name, $message_count, ) = $channel->queue_declare($this->queue,
+				false, true, false, false);
+
+			$data = [
+				'message_total' => $message_count,
+				'queue_name' => $q_name,
+				'message_add' => [
+					'mid' => $this->_mid,
+					'body' => $this->_body
+				]
+			];
+			if (($get = $channel->basic_get($q_name)) !== null) {
+				$data['basic_get'] = [
+					'mid' => $get->get('message_id'),
+					'body' => $get->body
+				];
+			}
+			$channel->close();
+			$connect->close();
 
-			Cache::setData($corrId, Cache::STATUS_SEND_OK);	//set: status = send ok
+			$statusKey = KeyHelper::getMessageStatusKey($this->_mid, $q_name);
+			Yii::$app->redis->set($statusKey, AmqpConfig::STATUS_HAND_OK);
 
-			$data[] = $corrId;
+			return $data;
 
-		} catch (\common\logic\Amqp\Exception $e) {
-			throw new Exception($e->getCode(), $e->getMessage());
+		} catch (\Exception $e) {
+			throw new Exception(1000, $e->getMessage());
+		}
+	}
+
+	/**
+	 * [构造消息体]
+	 * @author: libingke
+	 * @param $data
+	 */
+	private function _handleMessage($data)
+	{
+		$this->_mid = KeyHelper::getUniqueId('message_send');
+		$this->_body = call_user_func_array([$this, '_messageBody'], [$data]);
+		$properties = [
+			'message_id' 	=> $this->_mid,
+			'correlation_id'=> $this->_mid,
+			'consumer_tag'	=> $this->_mid
+		];
+		$this->_message = new AMQPMessage($this->_body, $properties);
+	}
+
+	private function _messageBody($data)
+	{
+		return json_encode($data);
+	}
+
+	/**
+	 * [批量发送消息]
+	 * @author: libingke
+	 * @return array
+	 * @throws Exception
+	 */
+	public function batchSendMessage()
+	{
+		try {
+			//connect
+			$connect = $this->getConnect();
+			$channel = $connect->channel();
+			//预声明
+			$channel->queue_declare($this->queue,
+				false, true, false, false);
+			//batch_basic_publish todo
+			foreach ($this->message as $k => $v) {
+				$this->_handleMessage($v);
+				$this->_rows[] = [
+					'mid' => $this->_mid,
+					'body' => $this->_body
+				];
+				$channel->batch_basic_publish($this->_message, '', $this->queue);
+
+				$statusKey = KeyHelper::getMessageStatusKey($this->_mid, $this->queue);
+				Yii::$app->redis->set($statusKey, AmqpConfig::STATUS_HAND_OK);
+			}
+			$channel->publish_batch();
+
+			//获取返回结果
+			list($q_name, $message_count, ) = $channel->queue_declare($this->queue,
+				false, true, false, false);
+
+			$data = [
+				'message_total' => $message_count,
+				'queue_name' => $q_name,
+				'add_count' => count($this->_rows),
+				'add_rows' => $this->_rows
+			];
+			if (($get = $channel->basic_get($q_name)) !== null) {
+				$data['basic_get'] = [
+					'mid' => $get->get('message_id'),
+					'body' => $get->body
+				];
+			}
+			$channel->close();
+			$connect->close();
+
+			return $data;
+
+		} catch (\Exception $e) {
+			throw new Exception(1000, $e->getMessage());
+		}
+	}
+
+	/**
+	 * [获取消息列表]
+	 * @author: libingke
+	 */
+	public function getMessageList()
+	{
+		/* api错误码 */
+		$badCode = 2000;
+		/* 登录验证 */
+		$authStr = Yii::$app->Amqp->getConfig('user') . ':' . Yii::$app->Amqp->getConfig('pass');
+		/* URL */
+		$vhost = urlencode(Yii::$app->Amqp->getConfig('vhost'));
+		$url = Yii::$app->Amqp->getConfig('host') . ':' . Yii::$app->Amqp->getConfig('api_port') .
+			"/api/queues/{$vhost}/" . $this->name . "/get";
+		$postParams = [
+			'name' => $this->name,
+			'count' => $this->count,
+			'encoding' => $this->encoding,
+			'requeue' => $this->requeue,
+			'truncate' => "50000",
+			'vhost' => '/',
+		];
+
+		$curl = new Curl();
+		$curl->setOption(CURLOPT_USERPWD, $authStr);
+		$curl->setRawPostData(json_encode($postParams));
+		$result = json_decode($curl->post($url), true);
+		if ($curl->responseCode != 200)
+			throw new Exception($badCode);
+		if ($curl->errorText)
+			throw new Exception($badCode, $curl->errorText);
+		if (isset($result['error']) && is_string($result['error']))
+			throw new Exception($badCode, $result['error']);
+
+		ArrayHelper::multisort($result,'message_count',SORT_ASC);
+		//print_r($result);exit();
+		$rows = [];
+		foreach ($result as $k => $v) {
+			$rows[$k]['before_count']	= $v['message_count'];
+			$rows[$k]['payload']			= $v['payload'];
+			$rows[$k]['payload_encoding']= $v['payload_encoding'];
+			$rows[$k]['message_id']		= $v['properties']['message_id'];
+		}
+		unset($result);
+
+		return ['count' => count($rows), 'rows' => $rows];
+	}
+
+	/**
+	 * [空处理]
+	 */
+	private function _consumeEmpty($mid, $body, $ack, $error, $stop = false)
+	{
+		$this->_stop = $stop;
+		$this->_result[] = [
+			'mid' => $mid,
+			'result' => 'success: do nothing!',
+			'ack' => $ack,
+			'error' => $error
+		];
+	}
+
+	/**
+	 * [处理逻辑1]
+	 * @author: libingke
+	 */
+	private function _consumeLogin($mid, $body, $ack, $error, $stop = false)
+	{
+		$this->_stop = $stop;
+		$this->_result[] = [
+			'mid' => $mid,
+			'result' => '已处理',
+			'ack' => $ack,
+			'error' => $error
+		];
+	}
+
+	/**
+	 * [消费某条消息]
+	 * @author: libingke
+	 * @param $mid
+	 * @return array
+	 * @throws Exception
+	 */
+	protected function consumeByMid($mid)
+	{
+		$q_name = $this->name;
+
+		//选择执行回调
+		if ($this->do_nothing != true) {
+			$function = '_consume' . ucfirst($q_name);
+			if (!method_exists($this, $function) || !is_callable(array($this, $function)))
+				throw new Exception(2015);
+		} else {
+			//if do nothing handle empty
+			$function = '_consumeEmpty';
+		}
+
+		$connect = $this->getConnect();
+		$channel = $connect->channel();
+		$channel->queue_declare($q_name,
+			false, true, false, false);
+		list(, $count, ) = $channel->queue_declare($q_name,
+			false, true, false, false);
+		$callback = function ($msg) use($function, $mid, $q_name) {
+			try {
+				$message_id = $msg->get('message_id');
+				$ack = $this->ack == true ? true : false;//是否应答
+				if ($mid == $message_id) {
+					call_user_func_array(
+						[$this, $function],
+						[$message_id, $msg->body, $ack, '',	true]
+					);
+
+					$statusKey = KeyHelper::getMessageStatusKey($message_id, $q_name);
+					if ($ack === true) {
+						$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
+						Yii::$app->redis->set($statusKey, AmqpConfig::STATUS_HAND_OK);
+					} else {
+						Yii::$app->redis->set($statusKey, AmqpConfig::STATUS_NO_ACK);
+					}
+				}
+			} catch (\Exception $e) {
+				//$e->getMessage();
+			}
+		};
+
+		try {
+			$channel->basic_qos(0, $count, null);
+			$channel->basic_consume($q_name,
+				'', false, false, false, false, $callback);
+
+			$i = 0;
+			while (count($channel->callbacks)) {
+				$i ++;
+				if ($i > $count || $this->_stop == true)
+					break;
+
+				$channel->wait();
+			}
+
+		} catch (\Exception $e) {
+			throw new Exception(2104, $e->getMessage());
+		}
+
+		$channel->close();
+		$connect->close();
+
+		if ($this->_result == null)
+			throw new Exception(2102);
+
+		return $this->_result;
+	}
+
+	/**
+	 * [根据数量消费]
+	 * @author: libingke
+	 * @param string | int $startPos 开始位置
+	 * @param int $count 数量
+	 */
+	protected function consumeByCount($count)
+	{
+		$q_name = $this->name;
+
+		//选择执行回调
+		if ($this->do_nothing != true) {
+			$function = '_consume' . ucfirst($q_name);
+			if (!method_exists($this, $function) || !is_callable(array($this, $function)))
+				throw new Exception(2015);
+		} else {
+			//if do nothing handle empty
+			$function = '_consumeEmpty';
+		}
+
+		$connect = $this->getConnect();
+		$channel = $connect->channel();
+		$channel->queue_declare($q_name,
+			false, true, false, false);
+		list(, $total, ) = $channel->queue_declare($q_name,
+			false, true, false, false);
+
+		$callback = function ($msg) use($function, $q_name) {
+			$ack = $this->ack == true ? true : false;//是否应答
+			try {
+				$message_id = $msg->get('message_id');
+				call_user_func_array(
+					[$this, $function],
+					[$message_id, $msg->body, $ack,	'', false]
+				);
+
+				//更新状态
+				$statusKey = KeyHelper::getMessageStatusKey($message_id, $q_name);
+				if ($ack === true) {
+					$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
+					Yii::$app->redis->set($statusKey, AmqpConfig::STATUS_HAND_OK);
+				} else {
+					Yii::$app->redis->set($statusKey, AmqpConfig::STATUS_NO_ACK);
+				}
+
+			} catch (\Exception $e) {
+				//消息体出错机制
+				call_user_func_array(
+					[$this, $function],
+					['', $msg->body, $ack, $e->getMessage(), false]
+				);
+			}
+		};
+
+		try {
+			$min = min($count, $total);
+			$channel->basic_qos(0, $min, null);
+			$channel->basic_consume($q_name,
+				'', false, false, false, false, $callback);
+
+			$i = 0;
+			while (count($channel->callbacks)) {
+				$i ++;
+				if ($i > $min || $this->_stop == true)
+					break;
+
+				$channel->wait();
+			}
+
+		} catch (\Exception $e) {
+			throw new Exception(2104, $e->getMessage());
+		}
+
+		$channel->close();
+		$connect->close();
+		return $this->_result;
+	}
+
+	/**
+	 * [消费消息]
+	 * @author: libingke
+	 */
+	public function consumeMessage()
+	{
+		switch ($this->type)
+		{
+			case static::TYPE_MID:
+				$data = $this->consumeByMid($this->mid);
+				break;
+
+			case static::TYPE_COUNT:
+				$data = $this->consumeByCount($this->count);
+				break;
+
+			case static::TYPE_MC:
+				throw new Exception(1000, '未开发');
+				break;
+
+			default:
+				return "It's not possible to get there.";
 		}
 
 		return $data;
 	}
+
+	/**
+	 * [清空消息]
+	 * @author: libingke
+	 * @return array
+	 */
+	public function purge()
+	{
+		$data = [];
+		$connect  = Yii::$app->Amqp->AMQPConnection();
+		$channel  = new \AMQPChannel($connect);
+		$queue = new \AMQPQueue($channel);
+		$queue->setName($this->name);
+		$queue->purge();
+		return $data;
+	}
+
 }

backend/forms/QueueForm.php

@@ -0,0 +1,191 @@
+<?php
+
+namespace backend\forms;
+
+use backend\models\Queue;
+use common\helpers\KeyHelper;
+use components\Curl;
+use components\Exception;
+use Yii;
+
+/**
+ * Class QueueForm
+ * @package backend\forms
+ */
+class QueueForm extends BaseForm
+{
+	public $sign;
+
+	public $queue;
+
+	public $remark = '';
+
+	/**
+	 * @var array
+	 */
+	public $config = [];
+
+	public $qid;
+
+	public $force;//强制删除：用于删除库里没有的队列
+
+	public function rules()
+	{
+		return [
+			//create_queue
+			[['config', 'remark'], 'safe', 'on' => 'create_queue'],
+			[['sign', 'queue'], 'required', 'on' => 'create_queue'],
+			['sign', 'filter', 'filter' => 'trim', 'on' => 'create_queue'],
+			['sign', 'string', 'on' => 'create_queue'],
+			['remark', 'default', 'value' => 'sys', 'on' => 'create_queue'],
+
+			//delete_queue
+			[['qid'], 'required', 'on' => 'delete_queue'],
+		];
+	}
+
+	/**
+	 * [创建队列]
+	 * @author: libingke
+	 * @return array
+	 * @throws Exception
+	 */
+	public function createQueue()
+	{
+		try {
+			$connect = new \PhpAmqpLib\Connection\AMQPStreamConnection(
+				Yii::$app->Amqp->host,
+				Yii::$app->Amqp->port,
+				Yii::$app->Amqp->user,
+				Yii::$app->Amqp->pass,
+				Yii::$app->Amqp->vhost
+			);
+			$channel = $connect->channel();
+			list($queue,,) = $channel->queue_declare($this->queue,
+				false, true, false, false);
+
+		} catch (\PhpAmqpLib\Exception\AMQPProtocolChannelException $e) {
+			throw new Exception(2000, $e->getMessage());
+		}
+
+		$data = ['queue' => $queue];
+		try {
+			$one = Queue::findOne(['sign' => $this->sign]);
+			if ($one)
+				throw new Exception(2008);
+
+			//save
+			$one = new Queue();
+			$one->qid = KeyHelper::getUniqueId('queue-add');
+			$one->sign = $this->sign;
+			$one->queue = $this->queue;
+			$one->status = Queue::STATUS_YES;
+			$one->remark = $this->remark;
+			$one->config = serialize($this->config);
+			if ($one->save()) {
+				$data['qid'] = $one->qid;
+				$data['sign'] = $one->sign;
+				ksort($data);
+			}
+
+		} catch (\yii\db\Exception $e) {
+			//todo db没有记录成功
+		}
+
+		return $data;
+	}
+
+	/**
+	 * [删除队列]
+	 * @author: libingke
+	 * @return array
+	 * @throws Exception
+	 */
+	public function deleteQueue()
+	{
+		$cb = function($name) {
+			try {
+				$connect = new \PhpAmqpLib\Connection\AMQPStreamConnection(
+					Yii::$app->Amqp->host,
+					Yii::$app->Amqp->port,
+					Yii::$app->Amqp->user,
+					Yii::$app->Amqp->pass,
+					Yii::$app->Amqp->vhost
+				);
+				$channel = $connect->channel();
+				$message = $channel->queue_delete($name,
+					false, false, false, false
+				);
+
+			} catch (\PhpAmqpLib\Exception\AMQPProtocolChannelException $e) {
+				$message = $e->getMessage();
+			}
+			return $message == 0 ? '' : $message;
+		};
+
+		$data = ['delete_count' => 0, 'delete_message' => ''];
+		try {
+			$one = Queue::findOne(['qid' => $this->qid, 'status' => Queue::STATUS_YES]);
+
+			if ($one) {
+				$one->status = Queue::STATUS_NO;
+				if (($c = $cb($one->queue)) === '') {
+					$one->update();
+					$data['delete_count'] ++;
+				}
+
+				$data['delete_message'] = $c;
+			} else {
+				throw new Exception(2009);
+			}
+
+		} catch (\yii\db\Exception $e) {
+			$data['delete_message'] = $e->getMessage();
+		}
+
+		return $data;
+	}
+
+	/**
+	 * [获取消息列表]
+	 * @author: libingke
+	 */
+	public function getQueueList()
+	{
+		$badCode = 2000;
+		$authStr = Yii::$app->Amqp->getConfig('user') . ':' . Yii::$app->Amqp->getConfig('pass');
+		/* URL */
+		$vhost = urlencode(Yii::$app->Amqp->getConfig('vhost'));
+		$url = Yii::$app->Amqp->getConfig('host') . ':' . Yii::$app->Amqp->getConfig('api_port') . "/api/queues";
+
+		$curl = new Curl();
+		$curl->setOption(CURLOPT_USERPWD, $authStr);
+		$result = json_decode($curl->get($url), true);
+		if ($curl->responseCode != 200)
+			throw new Exception($badCode);
+		if ($curl->errorText)
+			throw new Exception($badCode, $curl->errorText);
+		if (isset($result['error']) && is_string($result['error']))
+			throw new Exception($badCode, $result['error']);
+
+		$rows = [];
+		foreach ($result as $k => $v) {
+			//$name = $v['name'];
+			$name = $v['name'];
+			$rows[$name]['name'] 			= $name;
+			$rows[$name]['messages_count']	= $v['messages'];
+			$rows[$name]['message_bytes']	= $v['message_bytes'];
+			$rows[$name]['messages_ready']	= $v['messages_ready'];
+			//$rows[$name]['message_stats']	= $v['message_stats'];
+			$rows[$name]['consumers_count']	= $v['consumers'];
+			$rows[$name]['auto_delete']		= $v['auto_delete'];
+			$rows[$name]['durable']			= $v['durable'];
+			//$rows[$name]['arguments']		= $v['arguments'];
+			$rows[$name]['state']			= $v['state'];
+			$rows[$name]['idle_since']		= $v['idle_since'];
+		}
+		unset($result);
+
+		return ['count' => count($rows), 'rows' => $rows];
+	}
+}

backend/models/Jobs.php

@@ -1,53 +0,0 @@
-<?php
-
-namespace backend\models;
-
-use Yii;
-
-/**
- * 配置队列管理
- * Class Jobs
- * @package app\models
- * @author libingke
- */
-class Jobs extends \yii\db\ActiveRecord
-{
-	const STATUS_RUN = 1;
-
-	const STATUS_STOP = 0;
-
-    /**
-     * @inheritdoc
-     */
-    public static function tableName()
-    {
-        return 'amqp_jobs';
-    }
-
-	/**
-	 * [获取场景]
-	 * @author: libingke
-	 * @param $sign
-	 * @return false|string
-	 */
-    public static function fetchDataBySign($sign, $useCache = true)
-	{
-		if (!is_string($sign) || $sign === '')
-			return false;
-
-		$cache = Yii::$app->redis;
-		$key = 'table:amqp_jobs:sign:' . $sign;
-		$data = $cache->get($key);
-
-		if ($data === false || $useCache == false) {
-			$info = self::find()
-				->where(['sign' => $sign, 'status' => static::STATUS_RUN])
-				->asArray()->one();
-			$data = serialize($info);
-
-			$cache->set($key, $data, 60);
-		}
-
-		return unserialize($data);
-	}
-}

backend/models/Queue.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace backend\models;
+
+use yii\behaviors\TimestampBehavior;
+use yii\db\Expression;
+
+/**
+ * 队列
+ * Class Queue
+ * @package backend\models
+ */
+class Queue extends \yii\db\ActiveRecord
+{
+	const STATUS_YES = 1;
+
+	const STATUS_NO = 0;
+
+    /**
+     * @inheritdoc
+     */
+    public static function tableName()
+    {
+        return '{{%queue}}';
+    }
+
+	/**
+	 * @inheritdoc
+	 */
+	public function behaviors()
+	{
+		return [
+			[
+				'class' => TimestampBehavior::className(),
+				'createdAtAttribute' => 'created_at',
+				'updatedAtAttribute' => 'updated_at',
+				//'value'   => new Expression('NOW()'),
+				'value'   => function() { return time();},
+			],
+		];
+	}
+}

backend/models/WorkerScript.php

@@ -0,0 +1,36 @@
+<?php
+
+namespace backend\models;
+
+/**
+ * worker
+ * Class WorkerScript
+ * @package backend\models
+ */
+class WorkerScript extends \yii\db\ActiveRecord
+{
+	const STATUS_RUN = 1;
+
+	const STATUS_STOP = 0;
+
+    /**
+     * @inheritdoc
+     */
+    public static function tableName()
+    {
+        return '{{%worker_script}}';
+    }
+
+	/**
+	 * [所有需要运行的脚本]
+	 * @author: libingke
+	 * @return array|\yii\db\ActiveRecord[]
+	 */
+    public static function getRunScriptList()
+	{
+		$data = self::find()
+			->where(['status' => static::STATUS_RUN])
+			->asArray()->all();
+		return $data;
+	}
+}

common/config/main.php

@@ -8,6 +8,15 @@ return [
 	'sourceLanguage' => 'en',
 	'language' => 'zh-CN',
     'components' => [
+		//  新添加的
+		'Amqp' => [
+			'class' => 'components\service\Amqp',
+			'host'	=> '127.0.0.1',
+			'port'	=> '5672',
+			'user'	=> 'guest',
+			'pass'	=> 'Guest',
+		],
+
         'cache' => [
             'class' => 'yii\caching\FileCache',
         ],
@@ -17,7 +26,7 @@ return [
             'hostname' => '127.0.0.1',
             'port' => 6378,
             'database' => 1,
-            'password' => 'redis12017',
+            'password' => 'redis12078',
         ],
 
 		'i18n' => [

common/helpers/Helper.php

@@ -1,21 +0,0 @@
-<?php
-namespace common\helpers;
-
-/**
- * Created by PhpStorm.
- * Author: libingke
- * Date: 2018/2/1
- */
-class Helper
-{
-	/**
-	 * [生成唯一Id]
-	 * @author: libingke
-	 * @return string
-	 */
-	public static function getUniqueId()
-	{
-		return md5(uniqid(md5(microtime(true)),true));
-	}
-}
-

common/helpers/KeyHelper.php

@@ -0,0 +1,40 @@
+<?php
+namespace common\helpers;
+
+/**
+ * Created by PhpStorm.
+ * Author: libingke
+ * Date: 2018/2/1
+ */
+class KeyHelper
+{
+	/**
+	 * [生成唯一Id]
+	 * @author: libingke
+	 * @return string
+	 */
+	public static function getUniqueId($addStr = '')
+	{
+		$sugar = is_string($addStr) && $addStr != '' ? $addStr : '';
+		return md5(uniqid(md5(microtime(true)),true) . $sugar);
+	}
+
+	public static function getMessageStatusKey($id, $queue)
+	{
+		if (is_string($queue) && is_string($id)) {
+			return "queue:{$queue}:message:{$id}:status";
+		}
+
+		return null;
+	}
+
+	public static function getMessageResultKey($id, $queue)
+	{
+		if (is_string($queue) && is_string($id)) {
+			return "queue:{$queue}:message:{$id}:result";
+		}
+
+		return null;
+	}
+}
+

common/logic/Amqp/Cache.php

@@ -10,7 +10,7 @@ use Yii;
  */
 class Cache
 {
-	const CACHE_HEADER	= 'amqp_v1_msg_';
+	const CACHE_HEADER	= 'message:';
 
 	const STATUS_SEND_FAIL	= -1;//发送失败
 	const STATUS_SEND_OK	= 1;//已发送,待处理

common/logic/Amqp/Message.php

@@ -2,7 +2,7 @@
 
 namespace common\logic\Amqp;
 
-use common\helpers\Helper;
+use common\helpers\KeyHelper;
 use PhpAmqpLib\Message\AMQPMessage;
 use Yii;
 
@@ -42,7 +42,7 @@ class Message extends Connect
 	 */
 	public function send($body, $routing_key)
 	{
-		$this->_corr_id = Helper::getUniqueId();
+		$this->_corr_id = KeyHelper::getUniqueId('message_send');
 		$properties = [
 			//'content_type' => 'text/plain',
 			'correlation_id' => $this->_corr_id,

common/messages/zh-CN/errorCode.php

@@ -76,12 +76,13 @@ return [
 	'599' => 'Network connect timeout error',
 
 
+	'1000' => 'Exception',
 	'1001' => '非POST提交',
 	'1002' => '非GET提交',
 	'1003' => '表单验证失败',
 
 	//message
-
+	'2000' => 'rabbitMQ连接失败',
 	'2001' => '消息主体不能为空',
 	'2002' => '消息格式错误',
 	'2003' => '队列名称不能为空',
@@ -89,11 +90,20 @@ return [
 	'2005' => 'data不能为空',
 	'2006' => 'data格式错误，必须是合法json',
 	'2007' => 'data格式错误，必须包含 \'body\',\'infos\'',
+	'2008' => '队列标识sign已存在',
+	'2009' => '队列id不存在或已删除',
+	'2010' => 'requeue只能是布尔值',
+	'2011' => 'encoding格式错误',
+	'2012' => '获取数量必须是整数',
+	'2013' => '获取数量太小，必须大于0',
+	'2014' => '获取数量太大，必须小于1000 [或联系开发解除限制]',
+	'2015' => '未找到该队列对应的消费回调',
+	'2016' => '消息id不能为空',
+	'2017' => '数量不能为空',
 
 	'2100' => '队列创建失败',
 	'2101' => '消息处理中',
 	'2102' => '未知消息',
-
-	'2201' => 'sign不能为空',
-	'2202' => '不支持的请求类型',
-];
+	'2103' => '消息发送至队列过程中失败',
+	'2104' => '队列不存在',
+];

components/ErrorHandler.php

@@ -54,7 +54,7 @@ class ErrorHandler extends baseErrorHandler
 		} elseif ($response->format === Response::FORMAT_RAW) {
 			$response->data = static::convertExceptionToString($exception);
 		} else {
-			$response->data = $this->convertExceptionToArray($exception);
+			$response->data = static::convertExceptionToArray($exception);
 		}
 		if ($exception instanceof HttpException) {
 			$response->setStatusCode($exception->statusCode);

components/service/Amqp.php

@@ -0,0 +1,64 @@
+<?php
+
+namespace components\service;
+
+use components\Exception;
+use PhpAmqpLib\Connection\AMQPStreamConnection;
+
+/**
+ * Amqp 服务组件
+ * Class Amqp
+ * @package components\service
+ */
+class Amqp extends AmqpConfig
+{
+	/* 以下即将废弃：因为使用跟踪不方便 */
+
+	/**
+	 * [获取连接]
+	 * @author: libingke
+	 * @return AMQPStreamConnection
+	 */
+	public function getConnect()
+	{
+		$conn = new \PhpAmqpLib\Connection\AMQPStreamConnection(
+			$this->host,
+			$this->port,
+			$this->user,
+			$this->pass,
+			$this->vhost
+		);
+		return $conn;//->channel()->queue_declare();
+	}
+
+	/**
+	 * [php拓展自带的连接]
+	 * @author: libingke
+	 * @return \AMQPConnection
+	 */
+	public function AMQPConnection()
+	{
+		$config = [
+			'host' => $this->host,
+			'port' => $this->port,
+			'login'=> $this->user,
+			'password' => $this->pass,
+			'vhost' => $this->vhost
+		];
+		$conn = new \AMQPConnection($config);
+		if (!$conn->connect())
+			throw new Exception(2000);
+
+		return $conn;
+	}
+
+	/**
+	 * @param string $key
+	 * @return mixed
+	 */
+	public function getConfig($key = '')
+	{
+		return is_string($key) && $key && isset($this->$key) ?
+			$this->$key : json_decode(json_encode($this), true);
+	}
+}

components/service/AmqpConfig.php

@@ -0,0 +1,47 @@
+<?php
+
+namespace components\service;
+
+class AmqpConfig
+{
+	public $host = '127.0.0.1';
+
+	public $port = 5672;
+
+	public $user = 'guest';
+
+	public $pass = 'Guest';
+
+	public $vhost = '/';
+
+	public $api_port = '15672';
+
+	/* status */
+	const STATUS_SEND_FAIL	= -1;//发送失败
+	const STATUS_SEND_OK	= 1;//已发送,待处理
+	const STATUS_WAIT		= 2;//等待中
+	const STATUS_HAND		= 3;//处理中
+	const STATUS_NO_ACK		= 4;//已处理，未应答
+	const STATUS_HAND_OK	= 99;//处理成功
+	const STATUS_HAND_FAIL	= -99;//处理失败
+
+	public static $statusMark = [
+		-1	=> '发送失败',
+		0	=> '未知状态',
+		1	=> '已发送，待处理',
+		2	=> '等待中',
+		3	=> '处理中',
+		4	=> '已处理，未应答',
+		99	=> '处理成功',
+		-99 => '处理失败',
+	];
+
+	public static function getMarkById($id)
+	{
+		if (isset(self::$statusMark[$id]) && is_numeric($id)) {
+			return self::$statusMark[$id];
+		}
+
+		return self::$statusMark[0];
+	}
+}

console/controllers/MainController.php

@@ -1,6 +1,7 @@
 <?php
 namespace console\controllers;
 
+use backend\models\WorkerScript;
 use yii\console\Controller;
 
 /**
@@ -14,15 +15,7 @@ use yii\console\Controller;
  */
 class MainController extends Controller
 {
-	const DURATION = 10;	//主进程检查间隔时间
-
-	/**
-	 * 子进程
-	 * @var array ['route' => '', 'params' => '']
-	 */
-	private $_task = [
-		['route' => 'worker-msg/login', 'params' => '']
-	];
+	const DURATION = 30;	//主进程检查间隔时间
 
 	/**
 	 * [init]
@@ -47,9 +40,15 @@ class MainController extends Controller
 	public function checkTasks()
 	{
 		echo "\n Checking tasks......";
-		$list = $this->_task;
+		$list = WorkerScript::getRunScriptList();
 		foreach ($list as $task) {
-			$this->execTask($task['route']);
+			if (!is_string($task['controller']) || !is_string($task['action'])) {
+				echo "\n this task has error!";
+				continue;
+			}
+
+			$program = $task['controller'] . '/' . $task['action'];
+			$this->execTask($program);
 		}
 	}
 

console/controllers/WorkerMsgController.php

@@ -2,11 +2,11 @@
 
 namespace console\controllers;
 
-use backend\forms\MessageForm;
-use backend\models\Jobs;
+use backend\models\Queue;
 use common\logic\Amqp\Cache;
 use common\logic\Amqp\Connect;
 use components\Curl;
+use Yii;
 
 /**
  * 消息处理进程
@@ -41,12 +41,14 @@ class WorkerMsgController extends BaseController
 	 */
     public function actionLogin()
     {
-    	$queue = Jobs::find()->select('queue')->where(['sign' => 'queue'])->scalar();
+    	$queue = Queue::find()->select('queue')->where(['sign' => 'licai_login'])->scalar();
+    	if (!$queue)
+    		exit("no login queue\r\n");
 
     	$conn = $this->getConn();
 		$channel = $conn->channel();
 
-        $callback = function ($msg) {
+        $callback = function ($msg) use($queue) {
         	$cacheTime = 600;
 			$status = Cache::STATUS_HAND_FAIL;
 			$corr_id = $msg->get('correlation_id');
@@ -57,29 +59,36 @@ class WorkerMsgController extends BaseController
 				$data = ['code' => Cache::STATUS_HAND_FAIL, 'message' => '消息格式错误'];
 			} else {
 				//handle
-				$loginUrl = 'https://dev407.33.cn/admin/member/login';
-				$post['redirect_uri'] = 'https://zpapi.licai.cn';
-				$curl = new Curl();
-				$curl->setPostParams($post);
-				$result = json_decode($curl->post($loginUrl), true);
-
-				if (is_array($result) && isset($result['code'])) {
-					$cacheTime = isset($result['data']['expires_in']) ? $result['data']['expires_in'] : 864000;
-					$status = Cache::STATUS_HAND_OK;
-					$data = $result;
+				if (isset($post['url'])) {
+					$url = $post['url'];
+					unset($post['url']);
+					$post['redirect_uri'] = 'https://zpapi.licai.cn';
+
+					$curl = new Curl();
+					$curl->setPostParams($post);
+					$result = json_decode($curl->post($url), true);
+
+					if (is_array($result) && isset($result['code'])) {
+						$cacheTime = isset($result['data']['expires_in']) ? $result['data']['expires_in'] : 600;
+						$status = Cache::STATUS_HAND_OK;
+						$data = $result;
+
+					} else {
+						$data = ['code' => Cache::STATUS_HAND_FAIL, 'message' => $curl->errorText];
+					}
+
+					echo "Done: " . $corr_id, "\n";
+					$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
 				} else {
-					$data = ['code' => Cache::STATUS_HAND_FAIL, 'message' => $curl->errorText];
+					$data = ['code' => Cache::STATUS_HAND_FAIL, 'message' => '消息格式错误[url error]'];
 				}
-
-				echo "Done: " . $corr_id, "\n";
-				$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
 			}
 
-			Cache::setData($corr_id, $status);
-			Cache::setData('result_' . $corr_id, $data, $cacheTime);
+			Cache::setData("{$queue}:mid:{$corr_id}", $status);
+			Cache::setData("result:{$queue}:mid:{$corr_id}", json_encode($data), $cacheTime);
         };
 
-        $channel->basic_qos(null, 1, null);
+        $channel->basic_qos(null, 2, null);
         $channel->basic_consume($queue, '', false, false, false, false, $callback);
 
         while (count($channel->callbacks)) {
@@ -89,4 +98,44 @@ class WorkerMsgController extends BaseController
         $channel->close();
         $conn->close();
     }
+
+    public function callback($msg)
+	{
+		echo "$msg---" . PHP_EOL;
+	}
+
+	public function actionTest()
+	{
+		$connect = new \PhpAmqpLib\Connection\AMQPStreamConnection(
+			'121.196.226.188',
+			'5672',
+			'lbk',
+			'123456',
+			Yii::$app->Amqp->vhost
+		);
+		$channel = $connect->channel();
+		$channel->queue_declare('login', false, true, false, false);
+
+		$callback = function ($msg) {
+			call_user_func_array([$this, 'callback'], ['22']);
+			//$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
+		};
+
+		$channel->basic_qos(0, 10, null);
+		$channel->basic_consume('login', '', false, false, false, false, $callback);
+
+		$i = 0;
+		while (count($channel->callbacks)) {
+			$channel->wait();
+
+			$i++;
+			echo $i . PHP_EOL;
+			list($q_name, $message_count, $t) = $channel->queue_declare('login',
+				false, true, false, false);
+			var_dump($message_count, $t);
+		}
+
+		$channel->close();
+		$connect->close();
+	}
 }