消息回调

biz/message/services/InformAdminService.php

@@ -74,6 +74,7 @@ class InformAdminService extends BaseService
 			'createTime' => date("Y-m-d H:i:s"),
 		];
 		InformAdminClass::addData($informData);
+		return true;
 	}
 	
 }

common/components/rabbitmq/consumer/InformConsumer.php

@@ -15,35 +15,29 @@ use PhpAmqpLib\Message\AMQPMessage;
 
 class InformConsumer implements ConsumerInterface
 {
+	
 	/**
-	 * @param AMQPMessage $msg
-	 * @return bool
+	 * ConsumerInterface::MSG_ACK - Acknowledge message (mark as processed) and drop it from the queue
+	 * ConsumerInterface::MSG_REJECT - Reject and drop message from the queue
+	 * ConsumerInterface::MSG_REQUEUE - Reject and requeue message in RabbitMQ
 	 */
 	public function execute(AMQPMessage $msg)
 	{
 		$data = unserialize($msg->body);
-		//inform
-		//$messageType = $data['messageType'];
 		$userType = $data['userType'];
 		if ($userType == 'user') {
-			InformUserService::consumeInformQueue($data);
+			$return = InformUserService::consumeInformQueue($data);
+			if ($return) {
+				return ConsumerInterface::MSG_ACK;
+			}
 		}
 		if ($userType == 'admin') {
-			InformAdminService::consumeInformQueue($data);
-		}
-		if ($data['status'] == 200) {
-			print_r('suceess ');
-			return ConsumerInterface::MSG_ACK;
-		} elseif ($data['status'] == 500) {
-			sleep(5);
-			print_r('faile ');
-			return ConsumerInterface::MSG_REQUEUE;
+			$return = InformAdminService::consumeInformQueue($data);
+			if ($return) {
+				return ConsumerInterface::MSG_ACK;
+			}
 		}
-		/**
-		 * ConsumerInterface::MSG_ACK - Acknowledge message (mark as processed) and drop it from the queue
-		 * ConsumerInterface::MSG_REJECT - Reject and drop message from the queue
-		 * ConsumerInterface::MSG_REQUEUE - Reject and requeue message in RabbitMQ
-		 */
+		return ConsumerInterface::MSG_REQUEUE;
 	}
 	
 }