三.JMS API简析

四. JMS-API详解部分

1. ConnectionFactory接口:

    链接工厂,用于创建"链接",此处所指的连接为底层实际物理连接,即TCP连接(Socket通道).此接口有多个子接口:QueueConnectionFactory(用来创建Queue消息类型的链接),TopicConnectFactory,XAQueueConnectionFactory(基于XA分布式事务的Queue链接),XATopicConnectionFactory.

    ConnectionFactory接口中并没有约束建立链接所使用的协议、URL、安全策略等；这一切就交给JMS Provider去实现。通常情况下ConnectionFactory实例为单例,而且推荐以JNDI的方式获取.

Connection createConnection()Connection createConnection(String username,String password):使用简单的密码校验方式来创建链接.如果JMS Server端对受托管的queue/topic配置了需要授权才能访问,那么在建立相应的链接时需要交付密码.

2. Connection接口:

    代表底层一个物理(或者逻辑上)的一个Socket链接通道,此链接将会保持活跃直到JMS Client关闭或者JMS提供者(即JMS server端)在socket上阻塞超时.通常情况下,Connection为一个TCP链接(长连接),一个JMS Client建议维持一个Connection即可,事实上Queue/Topic不同类型的Client,那么也将创建不同的Connection.对于Server而言,Socket的资源开支是昂贵的,尽量避免一个Client创建多个Connection的情况.

    消息将通过"数据格式协议"在socket通道中传输,同一个connection中消息的发送/接受是有顺序的,这受限于Socket本身对流数据操作的特性.

    JMS Provider会为每个Connection生成ID,用来监控链接、消息分组等；JMS server端将会维护当前所有存活的Connection列表。

void start(): "开启"消息接收,此后即可接收消息;不过对于Producer而言,无论链接处于何种状态,均可以发送消息;此方法主要对消费者有效.此操作对Connection上所有的session都有效.void stop():"终止"消息接收,此后将不能接收到消息;当链接重新被start之后,将仍然可以继续接收.void close():关闭链接,底层为直接关闭socket;与Connection有关的临时(temporary)目的地都将被删除(包括TemporaryQueue,TemporaryTopic),以及此Connection有关的Sessions/productor/consumer都将被关闭.JMS规范要求,如果close方法返回意味着connection中所有的send操作已经结束,消息接收的receive方法已经返回(或中断);close方法会导致事务中的session无法继续,可能会rollback.Session createSession(boolean transacted,int acknowledgeMode):创建会话,并指定此Session的事务性和消息确认模式.String getClientID():获得当前JMS Client的ID;每个Connection,对于JMS提供者而言,就被认为是一个Client,clientId用来表示全局中唯一的一个链接,有server端生成;不同的JMS提供者对此ID的生成策略有所不同.void setClientID():设置ClientID,此操作需要在建立Connection之后,未使用Connection进行任何实际操作之前(包括创建session)进行;否则将会抛出异常.因为clientID是JMS server用来唯一标记链接的,因此在全局中不能重复,如果尝试设定一个已有的ClientId,将会抛出InvalidClientIDException.不过此方法可能在某些JMS Provider上不被支持.void setExceptionListener(ExceptionListener listener):设定Connection失效异常监听器,当JMS Client检测到链接异常,比如链接异常断开,将会通知此listener,可以在listener中做一些日志记录/补救措施,比如重新建立链接/会话恢复等.ConnectionMetaData getMetaData():获取JMS Provider中有关的元数据信息,比如版本号等.

3. DeliveryMode接口:

    "消息传输模式",此属性可以在session中指定,也可以在发送消息时指定;用来标记此消息是否需要被持久化,对于JMS而言,支持2种(作为DeliveryMode的静态属性):

    1) PERSISTENT: 表示消息需要被持久化,对于JMS Provider而言,这种类型的消息将会被存储在磁盘上;以确保在server故障恢复后,消息仍然保留.

    2) NON_PERSISTENT:表示消息不需要持久化,消息有可能被优先存储在内存中,或者存储在磁盘上某个临时文件上;当server故障失效后,消息将不能被恢复.

4.Destination接口:

    用来表示一个虚拟通道,或者说"目的地",消息的发送或者接收,都需要指定的destination.常见的2个子接口为:Queue和Topic;根据Destination的约束条件不同,可以分为"受托管Destination"、"动态Destination"、“临时Destination”。

    其中"受托管Destination"为JMS Provider中通过配置的方式声明的,无法通过外部API直接修改,此destination的使用需要受到JMS server管理员的授权等.

    "动态Destination"为通过JMS API方式创建的,比如session.createQueue(String queueName);这种类型的destination可以给JMS Client使用者提供了更多的自由空间,更加常用.

    "临时Destination"相对于"durable"(耐久的),这种类型的destination只能被当前Client感知到,它的生命周期和使用范围局限于Connection;即只有创建它的session所属的Connection中的其他消费者或者生产者才能使用它.这种destination在某些场景下很有用.

5. ExceptionListener接口:

    主要用来处理connection级别异常,比如链接异常断开;你可以在此listener中增加比如"链接重建"/"会话恢复"等措施;但是对于业务异常,此listener将不负责管理.

void onMessage(JMSException exception)

6. Session接口:

    最常用接口,考虑到Connection的资源开支较大,那么JMS提供了API级别的逻辑上的"链接",即Session;它可以更小粒度的控制消息属性(比如,确认模式,事务支持)以及消息的发送和接收.

    session通常在单线程中使用,无论是MessageProducer还是Consumer;而且通常情况下,一个Session只维护一个Producer实例或者Consumer实例;此外session的创建是非常便捷的.

    通过Session,可以创建多种类型的Producer和Consumer,而且事务的支持,也被控制在session级别.因此在支持事务的情况下,多个consumer或者Producer公用一个Session实例是不明智的.当然这也不是错误,不过前提需要注意:session中数据的操作并非在多线程下,能够得到预期的效果.

    如果你期望一个Producer持续的send消息,而另一个Consumer能够通过listener的方式接收消息,那么你应该将它们放在两个session中.此外如果你的消费者是基于listener异步接收消息,那么你应该为每个listener使用不同的session.

    在支持事务的Session中,多个消息的发送或者接收作为一个原子性单元,当事务提交后,消息的"确认"也是

作为原子性单元(此处消息的确认包括事务中多个发送的消息,或者消费者中连续消费的多个消息);如果事务回滚,将导致此事务中发送的消息被销毁(JMS server端做删除操作),对于消费者而言,事务中接收的消息将会被恢复(即认为消息未被收到,将会被重发).由此可见,在事务类型的session中,消息的确认时机将和事务提交的时机保持一致.

    在事务类型的Session中,如果事务没有提交,那么生产者send的消息对其他消费者不可见;对于消费者而言,如果事务没有提交,那么消息将不会从queue中删除,对于topic类型,消息将不会从自己的"消息副本"中删除.

    session并没有start()方法,默认每次commit之后就会开启一个新的事务.

static int AUTO_ACKNOWLEDGE: 消息自动确认,即消费者从receive()方法成功返回时或者当messageListener.onMessage(..)方法成功返回时,进行消息确认.如果receive()方法内部或者onMessage方法内部抛出异常(未捕获),将会导致此消息不能被"确认";那么对于JMS Provider而言,则认为此消息消费失败,将会重发.(对于某些Provider而言,将会在重发多次后仍然失败,将会考虑将消息转发给其他Connection).static int CLIENT_ACKNOWLEDGE: 客户端确认，即消息的确认需要client端选择时机手动去触发。这个方式给消息的确认提供了更加自由的方式；此方式针对消息消费者，消费者可以在接收到消息后，在任意时间调用此message.acknowledge()方法来确认消息。static int DUPS_OK_ACKNOWLEDGE: "可重复消息确认"，此模式可以允许JMS提供者将一条消息向同一个目的地发送两次以上。void close(): 关闭session；直接导致与此session有关的资源被释放，如果消费者的recevie()正在接收消息(而不是wait阻塞)或者messageListener.onMessage()方法正在执行，那么session关闭将会被阻塞。session关闭将会导致尚未提交的事务被回滚。session关闭后，此session创建的Producer或者Consumer将无法继续工作。BytesMessage createByteMessage():MapMessage createMapMessage():创建一条消息void commit():提交事务。void rollback():事务回滚。void recover():此方法只会被JMS Provider调用，JMS提供者将会暂停消息的发送，将此前已经发送但没有“确认”消息标记为“redelivered”，然后按照消息的原始顺序，将消息发送给Client端(包括redelivered和新消息)。MessageProducer createProducer(Destination des): 创建一个消息生产者。MessageConsumer createConsumer(Destination des): 创建一个消息消费者。MessageConsumer createConsumer(Destination des,String selector): 指定消息选择器。MessageConsumer createConsumer(Destination des,String selector,boolean noLocal): 指明当前消费者是否可以接受"本地"消息.noLocal参数只对Topic类型的"目的地"有效,如果消息消费者和生产者有一个Connection创建(即它们具有同一个ClientID，或者说底层是一个TCP链接)，即使它们是在不同的session中，它们均被认为是“本地”。Queue createQueue(String queueName):创建一个“队列”类型的目的地(动态队列)。Topic createTopic(String tpoicName):创建一个“主题”类型的目的地(动态主题)。TopicSubscriber createDurableSubscriber(Topic topic,String name): 创建一个“耐久订阅者”，并指定订阅者的名称(name,需要全局唯一)；如果一个Client需要接收Topic的全部信息，即使当Client的链接失效时也需要JMS 提供者保存它“错过”的消息。考虑到JMS 提供者内部的机制(消息副本)，需要为此“耐久订阅者”指定一个名称，且名称不能改动，否则此后消息不能接收到。创建“耐久订阅者”有个必要的先决条件：ClientID必须一致，你在创建“durableSubscriber”时，需要显示的指定“connection.setClientID(xxx)”，且ClientID在每次启动时必须一样，否则将无法使用“耐久订阅者”。(You cannot create a durable subscriber without specifying a unique clientID on a Connection)。QueueBrowser createBrowser(Queue queue,String selector): 类似于创建一个Consumer，不过此时创建的是一个“Browser”，只能查看消息队列，但不能消费(也不能干扰消费)。TemporaryQueue createTemporaryQueue(): 创建一个“临时队列”,其生命周期为当前Connection；如果当前session或者Connection关闭，那么queue也将不可用；同时此Queue只能被当前Connection下的其他session使用，对于其他Connection，此Queue是不可见的。(不能跨Connection，同时此方法并没有指名参数;Don‘t understand null destinations)TemporaryTopic createTemporaryTopic(): 同上void unsubscribe(String name): 取消“耐久订阅者”，此后JMS Provider将不会对此订阅者保留消息副本。

7. MessageConsumer接口:

    消息消费者顶级接口,其子接口有QueueReceiver和TopicSubscriber;可以在Session接口中通过各种方式创建MessageConsumer.消息接收可以是同步的,也可以是异步的.这取决于编码方式.

String getMessageSelector(): 获取当前消息的"消息选择器";Session接口中,已经提供了基于选择器创建Consumer的方法,注意:消息选择器必须在创建Consumer时指定,且整个session期间将无法再次改动,当然MessageConsumer接口中也没有提供设置选择器的方法;这个和JMS Provider对消息选择器的使用机制有关(后端过滤机制),稍后有专题专门介绍"消息选择器"的原理.Message receive(): 以同步的方式接收消息,同步意味着"阻塞",当在connection活跃期间,当前"目的地"中没有消息push过来(对于Topic)或者没有侦听到消息(对于Queue,普遍采用polling策略),那么此方法将会阻塞,直到接收到消息,或者consumer被关闭(close方法),或者connection/session被关闭,此时将会返回null.前文已经提到,在事务类型的session中,此方法返回时表示此消息已经被确认.Message receive(long timeout): 已同步的方式接收消息,不过指定了方法阻塞的时间;如果在超时时仍未收到消息,将会返回null.Message receiveNoWait(): 以非阻塞的方式接收消息,不过此处的非阻塞,只是尝试去获取,如果此时有消息亟待接收,将会返回message.否则返回null..此方法在一定程度上要求consumer使用"轮询"的方式获取消息,比如在while循环中.void close(): 关闭消费者,此方法会阻塞,直到正在接收消息的receive方法返回,或者基于messageListener.onMessage()方法执行结束.如果receive方法尚未收到消息,则返回null.

8. MessageProducer接口:

    消息生产者的顶级接口,负责向指定的"目的地"发送消息.两个子接口:QueneSender和TopicPublisher

void setDisableMessageID(boolean value): 是否关闭"messageID"选项,不过首先声明并不是所有的JMS Provider都会此选项感兴趣,有些JMS Provider会忽略此选项,会对所有的消息都生成messageID.有些JMS Provider则反其道而行之,将会忽略所有的MessageID,以便减少此值带来的额外的开支(比如网络传输数据量).在很多场景下,我们需要跟踪消息(或者过滤消息,"请求应答"模式),那么MessageID对我们将会非常有效.JMS API规定,任何一个MessageID必须为全局唯一的,以便Client端可以使用MessageID作为某种检索/存储的主键.setDisableMessageTimestamp(): 是否关闭"messageTimestamp"选项,它的机制和messageID一样;messageTimestamp用来表示此消息被发送的时间戳(send方法执行时);此选项通常可以用来跟踪消息被创建的时间.void setDeliveryMode(int mode): 设置消息的"传输模式";此设置将直接影响此Producer下所有发送的消息.void setPriority( int priority): 设置消息的"权重",共0~9个级别,默认为4,其中0~4表示普通优先级,5~9表示高优先级;优先级越高,将会导致此消息被优先发送给消费者.此后我们会详解"消息权重"和系统设计.void setTimeToLive(long timeToLive): 设置消息的最大存活时间,毫秒数.默认为0表示永不过期;JMS Provider在发送消息时都会检测消息的"存活有效性",如果消息过期,将会直接删除,而不发送给消费者.其中mesageTimestamp + timeToLive最终表示为消息过期的时间.在JMS Provider将消息发送给Client时,将使用过期时间和server的本地时间比较,以决定其是否过期.void close(): 关闭消息生产者.void send(Message message): 发送消息 ,不过此时将会使用MessageProducer指定的priority/deliveryMode等.void send(Message message,int deliveryMode,int priority,long timeToLive); 发送消息,并为此消息指定"传输模式"和"消息权重".如果某条消息需要特定声明这些属性,那么你可以使用此方法.void send(Destinantion des,Message message): 向指定的目的地,发送消息;这个方法并不常用,通常用在"请求-应答"模式中:即消息消费者接收到消息之后,需要临时创建一个Producer并将"应答消息"发送到指定的"replyTo"地址.---稍后介绍JMS API中所关联的具体知识点.

JMS API