ActiveMQ 使用总结

关键词： ActiveMQ    Liferay                                         

下载：http://www.activemq.org/site/download.html

ActiveMQ FAQ（使用中遇到的问题，大多在这里可以找到）: http://goopen.org/confluence/display/ACTIVEMQ/FAQ

红眼web观察: http://www.uuki.com/blog/index.php?2005/11/30/64-activemq

ActiveMQ中的消息持久性: http://blog.iecn.net/blog/html/do-showone-tid-889.html

结合spring使用ActiveMQ进行异步消息调用: http://blog.iecn.net/blog/html/do-showone-tid-892.html

ActiveMQ中的安全机制: http://blog.iecn.net/blog/html/do-showone-tid-890.html

结合Spring2.0和ActiveMQ进行异步消息调用： http://blog.iecn.net/blog/html/do-showone-tid-1035.html

ActiveMQ（ver 3.2.1）的jndi支持：http://blog.donews.com/foxgem/archive/2006/05/09/861535.aspx

Liferay 使用activemq时，每次启动都会重新生成derby，无法重发因为服务停止时，滞留在消息队列的消息。首先就要配置activemq每次从同一个数据库存取,增加一个broker（详见《红眼web观察》）

   <bean id="broker" class="org.activemq.spring.BrokerFactoryBean"> 
          <property name="config" value="classpath:/activemq.xml"/>
   </bean>

但此时仍然无法保证因为网络问题、服务器繁忙、邮件服务器问题等引起的无法正确投递的问题。
因为session没有使用事务，在出错抛出异常时没有回滚。

QueueSession session = con.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

即使使用事务，并且出错回滚的话仍然有问题。假如是因网络问题无法连接到邮件服务器等大问题时，
回滚后会继续发送本条消息，此时因网络问题不可能一时解决，jms就持续不断的发送该消息，会导致CPU 100％。

这样看来事务回滚也不是好的方法，只有在出错时记录到自己的DB中，然后写一个job或是task，每隔一段时间对因错没有发送的消息重新发送，也就是将其放置重新到队列当中。
同样的问题也存在将消息放置到队列中的情况，无法保证放置时出错后消息的重发，也要在出错时将消息保存到自己的DB中。

activemq derby的脚本：
ALTER TABLE ACTIVEMQ_TXS DROP CONSTRAINT SQL070117092033990;

ALTER TABLE ACTIVEMQ_MSGS DROP CONSTRAINT SQL070117092033260;

ALTER TABLE ACTIVEMQ_ACKS DROP CONSTRAINT SQL070117092034180;

DROP INDEX SQL070117092033260;

DROP INDEX SQL070117092033990;

DROP INDEX SQL070117092034180;

DROP INDEX ACTIVEMQ_MSGS_CIDX;

DROP INDEX ACTIVEMQ_ACKS_CIDX;

DROP INDEX ACTIVEMQ_MSGS_MIDX;

DROP TABLE ACTIVEMQ_MSGS;

DROP TABLE ACTIVEMQ_ACKS;

DROP TABLE ACTIVEMQ_TXS;

CREATE TABLE ACTIVEMQ_MSGS (
ID INTEGER NOT NULL,
CONTAINER VARCHAR(250),
MSGID VARCHAR(250),
MSG BLOB(1048576),
EXPIRATION BIGINT,
SENT_TO_DEADLETTER CHAR(1)
);

CREATE TABLE ACTIVEMQ_ACKS (
SUB VARCHAR(250) NOT NULL,
CONTAINER VARCHAR(250) NOT NULL,
LAST_ACKED_ID INTEGER,
SE_ID INTEGER,
SE_CLIENT_ID VARCHAR(250),
SE_CONSUMER_NAME VARCHAR(250),
SE_SELECTOR VARCHAR(250)
);

CREATE TABLE ACTIVEMQ_TXS (
XID VARCHAR(250) NOT NULL
);

CREATE UNIQUE INDEX SQL070117092033260 ON ACTIVEMQ_MSGS (ID ASC);

CREATE UNIQUE INDEX SQL070117092033990 ON ACTIVEMQ_TXS (XID ASC);

CREATE UNIQUE INDEX SQL070117092034180 ON ACTIVEMQ_ACKS (SUB ASC, CONTAINER ASC);

CREATE INDEX ACTIVEMQ_MSGS_CIDX ON ACTIVEMQ_MSGS (CONTAINER ASC);

CREATE INDEX ACTIVEMQ_ACKS_CIDX ON ACTIVEMQ_ACKS (CONTAINER ASC);

CREATE INDEX ACTIVEMQ_MSGS_MIDX ON ACTIVEMQ_MSGS (MSGID ASC);

ALTER TABLE ACTIVEMQ_TXS ADD CONSTRAINT SQL070117092033990 PRIMARY KEY (XID);

ALTER TABLE ACTIVEMQ_MSGS ADD CONSTRAINT SQL070117092033260 PRIMARY KEY (ID);

ALTER TABLE ACTIVEMQ_ACKS ADD CONSTRAINT SQL070117092034180 PRIMARY KEY (SUB, CONTAINER);


注：用mysql 的话，表ACTIVEMQ_TXS的字段SUB和CONTAINER 长度改为150.详见《ActiveMQ中的消息持久性》

自己数据库存取消息的POJO:
public class MailMessage implements Serializable {
 
    private Long id;
    private Blob message;
    private Serializable serialiableMsg;
 
 
    public Serializable getSerialiableMsg() throws SQLException {
        InputStream is = getMessage().getBinaryStream();
        serialiableMsg = (is == null) ? null : (Serializable) SerializationUtils.deserialize(is);
        return serialiableMsg;
    }
 
    public void setSerialiableMsg(Serializable serialiableMsg) {
        this.serialiableMsg = serialiableMsg;
        byte[] b = SerializationUtils.serialize(serialiableMsg);
        setMessage(b == null ? null : Hibernate.createBlob(b));
    }
 
    public Long getId() {
        return id;
    }
    public void setId(Long id) {
        this.id = id;
    }
    protected Blob getMessage() {
        return message;
    }
    protected void setMessage(Blob message) {
        this.message = message;
    }