在使用rocketmq之前使用了rabbitmq，会出现丢消息的情况，进而果断放弃，继续投入到大Java的怀抱，不过也遇到了一些问题，这里总结一下：
使用过程中新加节点需要手动创建topic
消费者处理不合理，不能实现负载均衡
针对第二点：我之前一直使用的是pull方式，按顺序来消费，一旦程序重启则从头一个一个消费，显然这种效率很低，
而且因为一个代码问题，如果我在offsize = 0的情况获取不了数据，则min offsize不增长，这种情况导致如果数据隔天了则不能消费，这时候消费者相当于在空跑。ListmsgList = new ArrayList (); try { Set mqs = consumer.fetchSubscribeMessageQueues(queueName); for (MessageQueue mq : mqs) { System.out.printf("Consume from the queue: " + mq + "%n"); try { PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32); if (pullResult != null) { putMessageQueueOffset(mq, pullResult.getNextBeginOffset()); switch (pullResult.getPullStatus()) { case FOUND: if (pullResult.getMsgFoundList() != null && pullResult.getMsgFoundList().size() > 0) { for (MessageExt messageExt : pullResult.getMsgFoundList()) { msgList.add(new MessageVo(messageExt.getMsgId(), messageExt.getBody())); } } break; case NO_MATCHED_MSG: break; case NO_NEW_MSG: break; case OFFSET_ILLEGAL: break; default: break; } } } catch (Exception e) { e.printStackTrace(); }
使用这种方法可以稳定的pull出数据，但是这种情况效率很低。
使用多线程方式：
private Queue
> messageQueue = new LinkedBlockingQueue
>(); DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(groupName); consumer.setNamesrvAddr(url); scheduleService = new MQPullConsumerScheduleService(groupName); scheduleService.setMessageModel(MessageModel.CLUSTERING); scheduleService.setDefaultMQPullConsumer(consumer); List msgList = new ArrayList (); try { Set mqs = consumer.fetchSubscribeMessageQueues(queueName); for (MessageQueue mq : mqs) { System.out.printf("Consume from the queue: " + mq + "%n"); try { PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32); if (pullResult != null) { putMessageQueueOffset(mq, pullResult.getNextBeginOffset()); switch (pullResult.getPullStatus()) { case FOUND: if (pullResult.getMsgFoundList() != null && pullResult.getMsgFoundList().size() > 0) { for (MessageExt messageExt : pullResult.getMsgFoundList()) { msgList.add(new MessageVo(messageExt.getMsgId(), messageExt.getBody())); } } break; case NO_MATCHED_MSG: break; case NO_NEW_MSG: break; case OFFSET_ILLEGAL: break; default: break; } } } catch (Exception e) { e.printStackTrace(); } List msgList = new ArrayList (); try { Set mqs = consumer.fetchSubscribeMessageQueues(queueName); for (MessageQueue mq : mqs) { System.out.printf("Consume from the queue: " + mq + "%n"); try { PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32); if (pullResult != null) { putMessageQueueOffset(mq, pullResult.getNextBeginOffset()); switch (pullResult.getPullStatus()) { case FOUND: if (pullResult.getMsgFoundList() != null && pullResult.getMsgFoundList().size() > 0) { for (MessageExt messageExt : pullResult.getMsgFoundList()) { msgList.add(new MessageVo(messageExt.getMsgId(), messageExt.getBody())); } } break; case NO_MATCHED_MSG: break; case NO_NEW_MSG: break; case OFFSET_ILLEGAL: break; default: break; } } } catch (Exception e) { e.printStackTrace(); }
通过回调的方式来定时调用。这种方式是多线程来实现的。
使用这种方式不能部署两个程序，因为会导致groupName冲突