HDC调试需求开发（15万预算）,能者速来！>>>
各位好，想请教一个比较奇怪的问题。在项目中，通过mina接受远程telnet命令，然后将命令中的请求参数放到消息队列里面，由消费者端去异步处理。但是在测试过程中发现，第一次的telnet命令，mq消费者端日志显示确认完成，但是onMessage方法日志内并未打印，业务代码就没执行。但是第二次发送同一个命令之后，一切就正常了。
1. spring-mq配置文件 <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd"> <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"> <property name="host" value="${mq.host}" /> <property name="username" value="${mq.username}" /> <property name="password" value="${mq.password}" /> <property name="port" value="${mq.port}" /> <property name="requestedHeartBeat" value="${mq.requestedHeartbeat}"/> <property name="publisherConfirms" value="${mq.publisherConfirms}" /> <property name="publisherReturns" value="${mq.publisherReturns}" /> </bean> <rabbit:listener-container id="listenerContainer" acknowledge="manual" connection-factory="connectionFactory"> <rabbit:listener ref="workerListener" queues="${mq.queue.lane.name}" /> </rabbit:listener-container> <rabbit:admin connection-factory="connectionFactory" /> <rabbit:queue auto-delete="false" durable="true" exclusive="false" id="queue_lane" name="${mq.queue.lane.name}" /> <rabbit:queue auto-delete="false" durable="true" exclusive="false" id="queue_payment" name="${mq.queue.payment.name}" /> <bean id="messageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /> <rabbit:direct-exchange name="directExchange" auto-delete="false" durable="true" id="directExchange"> <rabbit:bindings> <rabbit:binding queue="queue_lane" key="${mq.queue.lane.name}" /> <rabbit:binding queue="queue_payment" key="${mq.queue.payment.name}" /> </rabbit:bindings> </rabbit:direct-exchange> <bean id="rabbitTempate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"> <property name="connectionFactory" ref="connectionFactory" /> <property name="confirmCallback" ref="callbackListener" /> <property name="returnCallback" ref="returnCallbackListener" /> </bean> </beans>
2. consumer端代码 @Component("workerListener") public class LaneWorkerListener implements ChannelAwareMessageListener { private static final Logger log = LoggerFactory.getLogger(LaneWorkerListener.class); @Autowired private LaneProcessor processor; public LaneWorkerListener() { } public void onMessage(Message message, Channel channel) throws Exception { try { Object receivedObject = SerializationUtils.deserialize(message.getBody()); log.info("消费者端接收到消息:{}, 消息properties:{} ", receivedObject, message.getMessageProperties()); // business operations. if (receivedObject instanceof ParamsWrapper) { ParamsWrapper wrapper = (ParamsWrapper) receivedObject; processor.process(SessionManager.getSession(wrapper.getSessionId()), wrapper.getParams()); } channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); // false只确认当前一个消息收到，true确认所有consumer获得的消息 } catch (Exception e) { log.error(String.format("消息确认失败,详细信息:", e)); if (message.getMessageProperties().getRedelivered()) { log.error("消息MessageId:{}已重复处理失败,拒绝再次接收...", message.getMessageProperties().getMessageId()); channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息 } else { log.error("消息MessageId:{}即将再次返回队列处理...", message.getMessageProperties().getMessageId()); channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // requeue为是否重新回到队列 } } } }
3.另外还有两个确认机制的监听器 @Component("callbackListener") public class LaneWorkerCallbackListener implements ConfirmCallback { private static final Logger log = Logger.getLogger(LaneWorkerCallbackListener.class); public void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info(String.format("确认消息完成...,{%s},{%s},{%s}", correlationData, ack, cause)); } } @Component("returnCallbackListener") public class LaneWorkerReturnedCallbackListener implements ReturnCallback { private static final Logger log = Logger.getLogger(LaneWorkerCallbackListener.class); public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { log.info(String.format("消息返回处理中...,{%s},{%s},{%s},{%s},{%s}", message, replyCode, replyText,exchange,routingKey)); } }

在mina中的消息发送:
1. mina spring 配置 <bean id="ioAcceptor" class="org.apache.mina.transport.socket.nio.NioSocketAcceptor" init-method="bind" destroy-method="destroy"> <property name="defaultLocalAddress" ref="defaultLocalAddress"/> <property name="handler" ref="serverIoHandler" /> <property name="filterChainBuilder" ref="filterChainBuilder" /> <property name="reuseAddress" value="true" /> </bean>
2. 主要的消息接受处理handler ServerIOHandler @Service public class ServerIoHandler extends IoHandlerAdapter { private static final Logger LOGGER = Logger.getLogger(ServerIoHandler.class); @Autowired private CacheManager cacheManager; @Autowired private ILaneMessageProducer producer; @Autowired private LaneQueueContainer queues; @Override public void sessionCreated(IoSession session) throws Exception { //为session设置唯一主键id，暂用毫秒时间戳策略。缓存session session.setAttributeIfAbsent(LaneConstants.MINA_SESSION_PRIMARY_KEY, System.currentTimeMillis()); Object sessionId = session.getAttribute(LaneConstants.MINA_SESSION_PRIMARY_KEY); SessionManager.addSession(Long.valueOf(sessionId.toString()), session); //并且存在缓存中 String address = ((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress(); cacheManager.setPerpetual(LaneConfig.getEnv().concat(LaneConstants.CACHE_LANE_SESSION_KEY), address); } @Override public void messageReceived(IoSession session, Object message) throws Exception { try { LOGGER.info(String.format("Start: [%s] -- address:[%s]",message.toString(),session.getRemoteAddress().toString())); // 解密消息 if (LaneConfig.isEncryptedMsg()) { message = AESUtils.decrypt(message.toString()); } JSONObject msg = parseParam(message.toString()); Long spid = Long.valueOf(session.getAttribute(LaneConstants.MINA_SESSION_PRIMARY_KEY).toString());//session id. //消息存入消息队列 producer.sendMessage(queues.getQueueLane(), new ParamsWrapper(spid, msg)); LOGGER.info(String.format(">>>>>>消息{%s}已被放入队列中.....时间:[%s]", message.toString(),DateUtils.getCurrentDateTimeMilliSecond(new Date()))); } catch (Exception e) { SocketResponse srs = new SocketResponse(); srs.setCode(SocketResponse.INTERNAL_SERVER_ERROR); srs.setMsg("内部错误"); session.write(LaneUtils.parseReturn(srs)); LOGGER.error("Exception:", e); } finally { message = null; } }
还请各位大侠多多指点。。。。谢谢！