A stomp client base on netty.

Subscribe:

public class StompSubscribeTest {
 @Test
 public void testSub() throws Exception {
 StompClientManager stompClientManager = new StompClientManager();
 stompClientManager.connect("stomp://username:password@host:port");
 stompClientManager.createConsumer("/topic/logon")
 .id("wqteam_test")
 .ackMode(AckMode.AUTO)
 .handler(new MessageHandler() {
 @Override
 public void onMessage(Message message) {
 System.out.println(message);
 }
 })
 .subscribe();
 LockSupport.park();
 }
}

Sender:

public class StompSenderTest {
 @Test
 public void testSend() throws Exception {
 StompClientManager stompClientManager = new StompClientManager();
 stompClientManager.connect("stomp://username:password@10.232.136.85:61613");
 MessageProducer producer = stompClientManager.createProducer("/topic/logon3");
 for (int i = 0; i < 100; i++) {
 if (i > 5) {
 StompTransaction tx = producer.begin();
 Thread.sleep(1000);
 tx.send(i + "风格的歌");
 tx.abort();
 } else {
 producer.send(i + "大苏打");
 }
 try {
 ReceiptFuture future = producer.sendW((i + "大苏打").getBytes("UTF-8"));
 future.await();
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
 LockSupport.park();
 }
}

For Spring:

<bean id="stompClientManager" class="com.ohun.stomp.StompClientManager"
 init-method="start" destroy-method="stop">
 <!--<property name="uri" 
 value="stomp://username:password@10.232.136.85:61613?connectTimeout=3000"/>
 两种配置方式是一样的,看个人喜好-->
 <property name="config">
 <bean class="com.ohun.stomp.common.StompConfig">
 <property name="host" value="${wangxin.mq.stomp.host}"/>
 <property name="port" value="${wangxin.mq.stomp.port}"/>
 <property name="login" value="${wangxin.stomp.username}"/>
 <property name="pass" value="${wangxin.stomp.password}"/>
 <property name="connectTimeout" value="3000"/>
 <property name="heartbeatX" value="600000"/>
 <property name="heartbeatY" value="100000"/>
 <property name="monitorPeriod" value="60"/>
 <property name="connectCountPreHost" value="1"/>
 </bean>
 </property>
</bean>
<bean id="wx2PubMsgListener" class="com.ohun.test.WX2PublicMsgListener"
 init-method="init" destroy-method="destroy">
 <property name="topic" value="/topic/pamsgfromwx"/>
 <property name="clientId" value="wxadmin"/>
</bean>

public class WX2PublicMsgListener implements MessageHandler {
 @Resource
 private StompClientManager stompClientManager;
 public void init() {
 this.executor = newExecutor();
 MessageConsumer consumer = stompClientManager.createConsumer(topic);
 consumer.id(clientId).executor(executor).handler(this).subscribe();
 }
 @Override
 public void onMessage(final Message message) {
 logger.error(message.getTextBody())
 }
 public void destroy() throws Exception {
 executor.shutdown();
 }
 private ThreadPoolExecutor newExecutor() {
 final ThreadFactory threadFactory = new BasicThreadFactory.Builder()
 .daemon(true).namingPattern("wx-2-pub-%d").build();
 return new ThreadPoolExecutor(2, poolSize, 5L, TimeUnit.MINUTES,
 new LinkedBlockingQueue<Runnable>(queueSize),
 threadFactory,
 new RejectedExecutionHandler() {
 @Override
 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
 logger.warn("one message task was rejected total="
 + rejectedCount.incrementAndGet()
 + ",poolStatus=" + poolStatus());
 }
 });
 }
 private final Logger logger = LoggerFactory.getLogger(WX2PublicMsgListener.class);
 private AtomicInteger rejectedCount = new AtomicInteger(0);
 private int poolSize = 10, queueSize = 100;
 private ThreadPoolExecutor executor;
 private String topic;
 private String clientId;
}