Shulou(Shulou.com)05/31 Report--

This article mainly introduces "what are activemq features". In daily operation, I believe many people have doubts about what activemq features are. The editor consulted all kinds of materials and sorted out simple and easy-to-use methods of operation. I hope it will be helpful for you to answer the doubts about "what are activemq features?" Next, please follow the editor to study!

Activemq features: subscribe to multiple destination with wildcards and publish multiple destionation with combination

Activemq supports destination hierarchies [topic and queen] for easy classification and management.

There are three wildcards:

. Used to separate paths

* used to match a section of the path

Used to match the path of any section

Opics:.

. For example: football.division.leeds. If leeds participates in two sports, Scccer and Rugby, for convenience, we hope to see the latest achievements of Leeds through a message consumer. At this time, wildcards will have a chance to show their talents.

. : used to separate elements in the destination name

*: used to match one element

>: match one or all trailing elements

So, for the above example, you can subscribe to this topic: *. Leeds

If you want to know all the scores in the division1 division, you can subscribe to this: soccer.division1.*

If you want to know the Rugby score: you can subscribe to this: rugby. >.

However, the wildcard is for consumers, and if you send a topic: rugby. >., this message will only be sent to the named rugby. >. Not all themes start with rugby.

Here is a way to enable message producers to send a message

Messages are sent to multiple destinations. By using composite destination.

It is useful to send the same message to different destinations. For example, an application that stores information will send a message to the queue.

At the same time, this message should also be broadcast to all monitoring systems. Typically, you do this by sending messages twice with two producer. Composite destination is used to solve this situation.

For example, if you create a Queue named store.order.backoffice,store.order.warehouse, two Queue will be sent at the same time.

Subscription information interpretation

PRICE. > Any price for any product on any exchange

PRICE.STOCK. > Any price for a stock on any exchange

PRICE.STOCK.NASDAQ.* Any stock price on NASDAQ

PRICE.STOCK.*.IBM Any IBM stock price on any exchange

Since version 5.5, you can customize the path delimiter:

.

At this point, FOO.BAR.* can be expressed as FOO/BAR/*

You can also define other symbol bit path delimiters through the pathSeparator property.

Public void subscribeToLeeds () throws JMSException {

String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory (brokerURI)

Connection connection = connectionFactory.createConnection ()

Connection.start ()

Session session = connection.createSession (false, Session.AUTO_ACKNOWLEDGE)

Topic allLeeds = session.createTopic ("*. * .Leeds")

MessageConsumer consumer = session.createConsumer (allLeeds)

Message result = consumer.receive ()

}

11.1.2 send one message to multiple destinations

Send the same message to different destination: send a [queen,opic] combination mode to the case column. The default combination destination is separated by

Listed as store.order.backoffice,store.order.warehouse

String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory (brokerURI)

Connection connection = connectionFactory.createConnection ()

Connection.start ()

Session session = connection.createSession (false, Session.AUTO_ACKNOWLEDGE)

Queue ordersDestination = session.createQueue ("store.orders, topic://store.orders")

MessageProducer producer = session.createProducer (ordersDestination)

Message order = session.createObjectMessage ()

Producer.send (order)

11.2 Notification Messa

To put it simply, it realizes the record tracking and notification of various operations on the broker of ActiveMQ.

Using this function, you can know in real time on broker

The connection was created or destroyed

Survivor or consumer added or removed

Topics or queues added or removed

There are messages sent and received

When are there slow consumers?

When is there a fast producer?

When and what news was discarded?

When is broker added to the cluster (master-slave or network connection)

This mechanism is an important supplement to the JMS protocol by ActiveMQ, and is also a part of the manageability of ActiveMQ based on JMS implementation. Basic settings for mutual coordination and interoperation of multiple ActiveMQ.

String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory (brokerURI)

Connection connection = connectionFactory.createConnection ()

Connection.start ()

Session session = connection.createSession (false, Session.AUTO_ACKNOWLEDGE)

Topic connectionAdvisory = org.apache.activemq.advisory.AdvisorySupport.CONNECTION_ADVISORY_TOPIC

MessageConsumer consumer = session.createConsumer (connectionAdvisory)

ActiveMQMessage message = (ActiveMQMessage) consumer.receive ()

DataStructure data = (DataStructure) message.getDataStructure ()

If (data.getDataStructureType ()) = = ConnectionInfo.DATA_STRUCTURE_TYPE) {

ConnectionInfo connectionInfo = (ConnectionInfo) data

System.out.println ("Connection started:" + connectionInfo)

} else if (data.getDataStructureType ()) = = RemoveInfo.DATA_STRUCTURE_TYPE) {

RemoveInfo removeInfo = (RemoveInfo) data

System.out.println ("Connection stopped:" + removeInfo.getObjectId ())

} else {

System.err.println ("Unknown message" + data)

}

Most advisor messages are complete for destiation, but the advisorysupport class has some ways to decide which advisorytopic to listen on, and you can also use wildcards-

String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory (brokerURI)

Connection connection = connectionFactory.createConnection ()

Connection.start ()

/ / Lets first create a Consumer to listen too

Session session = connection.createSession (false, Session.AUTO_ACKNOWLEDGE)

/ / Lets first create a Consumer to listen too

Queue queue = session.createQueue ("test.Queue")

MessageConsumer testConsumer = session.createConsumer (queue)

/ / so lets listen for the Consumer starting/stoping

Topic advisoryTopic = org.apache.activemq.advisory.AdvisorySupport.getConsumerAdvisoryTopic (queue)

MessageConsumer consumer = session.createConsumer (advisoryTopic)

Consumer.setMessageListener (new MessageListener () {

Public void onMessage (Message m) {

ActiveMQMessage message = (ActiveMQMessage) m

Try {

System.out.println ("Consumer Count =" + m.getStringProperty ("consumerCount"))

DataStructure data = (DataStructure) message.getDataStructure ()

If (data.getDataStructureType ()) = = ConsumerInfo.DATA_STRUCTURE_TYPE) {

ConsumerInfo consumerInfo = (ConsumerInfo) data

System.out.println ("Consumer started:" + consumerInfo)

} else if (data.getDataStructureType ()) = = RemoveInfo.DATA_STRUCTURE_TYPE) {

RemoveInfo removeInfo = (RemoveInfo) data

System.out.println ("Consumer stopped:" + removeInfo.getObjectId ())

} else {

System.err.println ("Unknown message" + data)

}

} catch (JMSException e) {

E.printStackTrace ()

}

}

});

TestConsumer.close ()

In ActiveMQ, topic is persistent only under persistent subscriptions (durablesubscription). When there are persistent subscriptions, each persistent subscriber is the equivalent of a persistent queue client, which receives all messages. There are two problems in this case:

1. The problem of consumer load balancing in the same application: a persistent subscription on the same application cannot use multiple consumer to share the message processing function. Because each will get all the information. Queue mode can solve this problem, broker

The client cannot send messages to multiple applications. Therefore, the jms specification itself does not have to publish subscriptions and group consumers.

two。 The problem of failover on the client side of the same application: because only a single persistent subscriber can be used, if this subscriber makes an error, the application will not be able to process messages, and the robustness of the system is not high. In order to solve these two problems, virtual is implemented in ActiveMQ.

The function of Topic. It's very easy to use. For a message publisher, it is a normal Topic with the name VirtualTopic. The beginning. For example, VirtualTopic.TEST. For the message receiver, it is a queue, and different applications use different prefixes as

The name of the queue, you can indicate your identity, and you can realize the consumer application packet. For example, Consumer.A.VirtualTopic.TEST indicates that it is the consumer side named A, while Consumer.B.VirtualTopic.TEST indicates that it is a client named B.

If you can consume this queue using multiple consumer in the same application, you can achieve the above two functions. Because different applications use different queue names (prefixes are different), all messages can be received in different applications. Each client is equivalent to a persistent subscription

Reader, and this client can use multiple consumers to share consumption tasks.

String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory (brokerURI)

Connection consumerConnection = connectionFactory.createConnection ()

ConsumerConnection.start ()

Session consumerSessionA = consumerConnection.createSession (false,Session.AUTO_ACKNOWLEDGE)

Queue consumerAQueue = consumerSessionA.createQueue ("Consumer.A.VirtualTopic.orders")

MessageConsumer consumerA = consumerSessionA.createConsumer (consumerAQueue)

Session consumerSessionB = consumerConnection.createSession (false,Session.AUTO_ACKNOWLEDGE)

Queue consumerBQueue = consumerSessionB.createQueue ("Consumer.B.VirtualTopic.orders")

MessageConsumer consumerB = consumerSessionB.createConsumer (consumerAQueue)

/ / setup the sender

Connection senderConnection = connectionFactory.createConnection ()

SenderConnection.start ()

Session senerSession = senderConnection.createSession (false, Session.AUTO_ACKNOWLEDGE)

Topic ordersDestination = senerSession.createTopic ("VirtualTopic.orders")

MessageProducer producer = senerSession.createProducer (ordersDestination)

Consumers with the same queue name will share all messages equally.

The message received from queue, message.getJMSDestination (). ToString () is topic://VirtualTopic.TEST, which is the original destination. The persistent property of the message is true, that is, each is equivalent to a persistent subscription.

The VirtualTopic feature has a master switch on broker. The useVirtualTopics property, which defaults to true, can be turned off by setting it to false.

When this feature is enabled and persistent storage is used, broker starts up to get the names of all destinations from the persistent storage, and if the name pattern matches Virtual Topics, add them to the system's Virtual Topics list.

Of course, there is no explicitly defined Virtual Topics, but it can also be used directly, and the system will automatically create the corresponding actual topic.

When a consumer accesses this VirtualTopics, the system automatically creates a persistent queue and distributes it to the specific queue each time the Topic receives a message.

Traceable consumers, valid only for Topic, can get messages before instance creation if consumer is traceable. Generally speaking, it is impossible for subscribers to get messages before instance creation because broker does not know it exists at all. For broker, if

A Topic channel is created and the publisher publishes the message (Publisher), then broker will save the published message in memory (non-persistent) or on disk (persistent), and the original message content will not be purged until all subscribers are consumers. So, retroactive,

Type subscribers can get these old messages that are not their own but are still saved on broker, just as we subscribe to a Feed and can get a list of old content immediately. If the subscriber is not durable (durable), it can get some recently released messages; if it is durable, it can get all the old messages in memory that have not been deleted. [the data forwarding model of Topic is described in detail below]

/ / set in destinationUrl. Default is false.

FeedTopic?consumer.retroactive=true

On the broker side, you can configure the current Topic to be "traceable" by default, but Topics does not save additional messages in this case, it just means that subscribers are traceable by default.

String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory (brokerURI)

Connection connection = connectionFactory.createConnection ()

Connection.start ()

Session session = connection.createSession (false, Session.AUTO_ACKNOWLEDGE)

Topic topic = session.createTopic ("soccer.division1.leeds?consumer.retroactive=true")

MessageConsumer consumer = session.createConsumer (topic)

Message result = consumer.receive ()

RedeliveryPolicy

The retransmission strategy used by consumer will trigger message retransmission when message processing fails on the client side (for example, onMessage method throws an exception, transaction rollback, etc.). For the broker side, messages that need to be retransmitted will be sent immediately (if the broker side uses asynchronous sending

And there are other messages in the send queue, then the retransmitted message may not be reached immediately to Consumer). We use this Policy to configure the maximum number of retransmissions, retransmission frequency, and so on. If your Consumer client is in a bad network environment, you can adjust the relevant parameters appropriately. Parameter list

See (RedeliveryPolicy)

/ / set in brokerUrl

Tcp://localhost:61616?jms.redeliveryPolicy.maximumRedeliveries=6

. RedeliveryPolicy

RedelieveryPolicy policy=connection.getRedelieveryPolicy ()

Policy.setInitialRedelieveryDelay (500)

Policy.setBackOffMultiplier (2)

Policy.setUseExponentialBackOff (true)

Policy.setMaximumRedelieveries (2)

The DLQ- Dead letter queue (Dead Letter Queue) is used to hold messages that have failed or expired.

The message is redelivered when the following occurs

A transacted session is used and rollback () is called.

A transacted session is closed before commit is called.

A session is using CLIENT_ACKNOWLEDGE and Session.recover () is called.

When a message is redelivered more than maximumRedeliveries (the default is 6 times, please refer to the link below for specific settings), a "Poison ack" will be sent to broker. The message is considered to be a poison pill, and broker will send this

The message is sent to DLQ for subsequent processing. The default dead letter queue is ActiveMQ.DLQ, and if not specified, dead letters will be sent to this queue. The default persistent message expires and will be sent to DLQ. Non-persistent messages will not be sent to DLQ through the configuration file (activemq.xml).

To adjust the dead letter sending policy

...

In an electronic system, it is possible to accept a variety of order information from different suppliers, different types of orders take different processes, in order to quickly process different orders to complete different business. Special definition of different routing information. The message is processed differently according to the routing information. If you use ActiveMQ, it is best to use apache-camel integration, so that different messages are automatically processed to different queues according to different processes.

At this point, the study of "what are the features of activemq" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.