我们到底能走多远系列（29）
扯淡：
“然后我俩各自一端/望着大河弯弯/终于敢放胆/嘻皮笑脸/面对/人生的难” --- 《山丘》
“迎着风/迎向远方的天空/路上也有艰难/也有那解脱/都走得从容” --- 《与你到永久》
“遇上冷风雨休太认真/自信满心里休理会讽刺与质问/笑骂由人洒脱地做人/少年人洒脱地做人/继续行洒脱地做人” ---《沉默是金》
主题：
使用JMS将共通模块分离出去，比如发短信模块，可以在远程的机器上跑customer，然后各个应用使用发短信功能是只要向远程机器发送msg即可。
类似于下图：
对于图中的Producer的实现都差不多，主要是选择什么样的Jms第三方实现。对于Customer我们不必关心.
比如下面的代码是HornetQ的Producer的样例代码：
public class JmsProducer implements ExceptionListener,FailureListener{
private final Logger logger = LoggerFactory.getLogger(JmsProducer.class);
private String queueName;
private String jmsHost;
private int jmsPort;
private ConnectionFactory cf;
private Queue queue;
private Connection queueConnection;
private Session queueSession;
private MessageProducer queueProducer;
public void init() throws Exception {
queue = HornetQJMSClient.createQueue(queueName);
Map<String, Object> connectionParams = new HashMap<String, Object>();
connectionParams.put(TransportConstants.PORT_PROP_NAME, jmsPort);
connectionParams.put(TransportConstants.HOST_PROP_NAME, jmsHost);
TransportConfiguration transportConfiguration = new TransportConfiguration(NettyConnectorFactory.class.getName(),
connectionParams);
HornetQConnectionFactory hornetQConnectionFactory = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, transportConfiguration);
hornetQConnectionFactory.setClientFailureCheckPeriod(60000);
hornetQConnectionFactory.setRetryInterval(2000); // 2 seconds for first retry
hornetQConnectionFactory.setRetryIntervalMultiplier(1.5); // 1.5 times loner betrween retrys
hornetQConnectionFactory.setMaxRetryInterval(20000); // Wait max 20 secs between retrys
hornetQConnectionFactory.setReconnectAttempts(-1); // Retry forever
hornetQConnectionFactory.setConnectionTTL(60000); //The default value for connection ttl is 60000ms
hornetQConnectionFactory.setClientFailureCheckPeriod(30000);//The default value for client failure check period is 30000ms
cf = (ConnectionFactory)hornetQConnectionFactory;
queueConnection = cf.createConnection();
queueSession = queueConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
queueProducer = queueSession.createProducer(queue);
queueProducer.setTimeToLive(6000000);//100分钟失效
queueProducer.setDisableMessageID(true);//关闭消息id
queueProducer.setDisableMessageTimestamp(true);//关闭消息的时间戳
logger.info("init JmsProducer of "+queueName);
//queueConnection.start();
}
public void reConnect(){
logger.info(queueName+" reConnect");
}
public void destroy() throws Exception {
logger.info("destroy JmsProducer of "+queueName);
if(queueSession != null){
queueSession.close();
queueSession = null;
}
if(queueConnection != null){
queueConnection.close();
queueConnection = null;
}
}
public String getQueueName() {
return queueName;
}
public void setQueueName(String queueName) {
this.queueName = queueName;
}
public String getJmsHost() {
return jmsHost;
}
public void setJmsHost(String jmsHost) {
this.jmsHost = jmsHost;
}
public int getJmsPort() {
return jmsPort;
}
public void setJmsPort(int jmsPort) {
this.jmsPort = jmsPort;
}
public Session getQueueSession() {
return queueSession;
}
public void sendTextMessage(final TextMessage textMessage) throws JMSException {
try {
queueProducer.send(textMessage);
} catch (Exception e) {
// TODO: handle exception
logger.error("on sendTextMessage Exception="+e.getMessage());
}
}
public void onException(JMSException jmsex) {
// TODO Auto-generated method stub
logger.warn("on JmsProducer Exception="+jmsex.getMessage());
}
public void connectionFailed(HornetQException hqex, boolean arg1) {
// TODO Auto-generated method stub
logger.error("on JmsProducer connectionFailed,arg1="+arg1+",Exception="+hqex.getMessage());
}
}
一般性的，我们利用Spring 把这个JmsProducer 注入进自己的业务类里去使用即可：
spring的bean配置：
<bean id="jmsCodeProducer" class="com.sz.lvban.biz.bo.util.JmsProducer" init-method="init">
<property name="jmsHost" value="${jms.send.code.host}" />
<property name="jmsPort" value="${jms.send.code.port}" />
<property name="queueName" value="${jms.send.code.queueName}" />
</bean>
@Autowired
private JmsProducer jmsCodeProducer;
某方法直接调用：
textMsg = jmsCodeProducer.getQueueSession().createTextMessage();
textMsg.setText(smsJson.toJSONString());
jmsCodeProducer.sendTextMessage(textMsg);
这样就实现了让服务器上的customer干活的工作了。
然后，我们发现spring默认注入jmsCodeProducer使用了单例的模式，这样一来我们就可能考虑多线程调用冲突的问题。然而我们不能避免jmsCodeProducer的单例，毕竟init-method="init" 的init方法有点消耗的。
所以就搞了下面的方案：（上图的题部分）
使用一个queue做中间站，只要保证单线程从queue中取数据，就能实现一条条向远程服务器发送jms消息。
下面是一个实现的例子：
我们先配置一个ApnsMsgSender，有他来控制queue的行为，包括插入，取出数据，以及发送jms消息。
<bean id="ApnsMsgSender" class="com.sz.wxassistant.biz.bo.util.ApnsMsgSender" init-method="sendMsg">
<property name="jmsApnsProducer" ref="jmsApnsProducer"></property>
</bean>
注意init-method="sendMsg" 启动时，我们就需要启动一个线程来监控queue。
结合下代码：在这里我们使用了LinkedBlockingQueue，关于ArrayBlockingQueue和LinkedBlockingQueue之间的取舍，我没有实际测试过。
public class ApnsMsgSender {
// private ArrayBlockingQueue<TextMessage> queue = new
// ArrayBlockingQueue<TextMessage>(1024);
private LinkedBlockingQueue<TextMessage> jmsQueue = new LinkedBlockingQueue<TextMessage>();
private JmsProducer jmsApnsProducer;
private ExecutorService pool;
private Logger log = LoggerFactory.getLogger("ApnsMsgSender");
/**
* 启动入口
*/
public void sendMsg() {
pool = Executors.newCachedThreadPool(new MyThreadFactory());
pool.submit(new JmsSender());
}
public boolean addJms(TextMessage msg) {
return jmsQueue.offer(msg);
}
public TextMessage getMsg() {
TextMessage msg = null;
try {
// 取msg 10秒超时设置
msg = jmsQueue.poll(10, TimeUnit.SECONDS);
} catch (InterruptedException interuptedE) {
log.warn("poll jms error" + interuptedE);
} catch (Exception e) {
log.error("poll jms get unknown error: ", e);
}
return msg;
}
public JmsProducer getJmsApnsProducer() {
return jmsApnsProducer;
}
public void setJmsApnsProducer(JmsProducer jmsApnsProducer) {
this.jmsApnsProducer = jmsApnsProducer;
}
public TextMessage genTextMessage() throws JMSException {
return jmsApnsProducer.getQueueSession().createTextMessage();
}
private class JmsSender implements Runnable {
public void run() {
while (true) {
try {
// 从queue中取msg
TextMessage msg = getMsg();
if (msg != null && msg instanceof TextMessage) {
// 发送
jmsApnsProducer.sendTextMessage(msg);
}
} catch (JMSException jmsE) {
log.error("send jms error: " + jmsE);
} catch (Exception e) {
log.error("get unknown error: ", e);
}
}
}
}
class MyThreadFactory implements ThreadFactory {
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
// 线程设置为后台进程
thread.setDaemon(true);
thread.setName("ApnsMsgSender");
return thread;
}
}
}
解释：
我们的这个线程做了什么？
1, getMsg()
2, sendTextMessage(msg)
很明了的实现....
注：代码中还使用了ThreadFactory 来封装了一下线程。
外界代码调用怎么搞？
1，addJms 就可以了
只负责向queue里放，这样再多的线程都没有关系了。
TextMessage msg = apnsMsgSender.genTextMessage();
msg.setText("I love los angeles ！");
apnsMsgSender.addJms(msg);
ok。到这里就实现了单线程发送jms消息的功能。
让我们继续前行
----------------------------------------------------------------------
努力不一定成功，但不努力肯定不会成功。
共勉。