You seem to be expecting your network of brokers to behave like a cluster.
ActiveMQ doesn't support clustering; the closest we come is master/slave
groups, but only one broker is active at a time so that's completely
different. In a network of brokers, each broker is an independent entity
that consumes messages from its peers when necessary to route messages from
the broker on which messages are produced to the brokers to which consumers
are attached. In a 2-broker network like yours where one client is on each
broker, messages will traverse both brokers on their way from producer to
consumer, as you can see.
Tim
On Dec 11, 2016 12:19 PM, "rockies" <lakshmi.chapar...@bd.com> wrote:
> I am using the java classes from ActiveMQ
> apache-activemq-5.14.1\examples\openwire\java Publisher and Listener.
>
> Modified Listener code to use Queue foo.bar on 61616 , and on Publisher
> queue foo.bar at 63616:
>
>
> Web Console at 61616 shows :
>
> Name Number Of Pending Messages Number Of Consumers Messages Enqueued
> Messages Dequeued
> foo.bar 0 0
> 10001 10001
>
>
> Web Console at 63616 also shows the same :
>
> foo.bar 0 0 10001 10001
>
> I see network connection from localhost: 61616 to localhost:63616
> (broker1Tobroker2) Does this mean they are successfully networked? How come
> messages are not distributed evenly between brokers?
>
>
>
>
> import javax.jms.*;
>
> class Listener {
>
> public static void main(String []args) throws JMSException {
>
> String user = env("ACTIVEMQ_USER", "admin");
> String password = env("ACTIVEMQ_PASSWORD", "admin");
> String host = env("ACTIVEMQ_HOST", "localhost");
> int port = Integer.parseInt(env("ACTIVEMQ_PORT", "61616"));
> String destination = arg(args, 0, "foo.bar");
>
> ActiveMQConnectionFactory factory = new
> ActiveMQConnectionFactory("tcp://" + host + ":" + port);
>
> Connection connection = factory.createConnection(user, password);
> connection.start();
> Session session = connection.createSession(false,
> Session.AUTO_ACKNOWLEDGE);
> // Destination dest = new ActiveMQTopic(destination);
> Destination dest = new ActiveMQQueue(destination);
>
> MessageConsumer consumer = session.createConsumer(dest);
> long start = System.currentTimeMillis();
> long count = 1;
> System.out.println("Waiting for messages...");
> while(true) {
> Message msg = consumer.receive();
> if( msg instanceof TextMessage ) {
> String body = ((TextMessage) msg).getText();
> if( "SHUTDOWN".equals(body)) {
> long diff = System.currentTimeMillis() - start;
> System.out.println(String.format("Received %d in %.2f
> seconds", count, (1.0*diff/1000.0)));
> break;
> } else {
> if( count != msg.getIntProperty("id") ) {
> System.out.println("mismatch:
> "+count+"!="+msg.getIntProperty("id"));
> }
> count = msg.getIntProperty("id");
>
> if( count == 0 ) {
> start = System.currentTimeMillis();
> }
> if( count % 1000 == 0 ) {
> System.out.println(String.format("Received %d
> messages.", count));
> }
> count ++;
> }
>
> } else {
> System.out.println("Unexpected message type:
> "+msg.getClass());
> }
> }
> connection.close();
> }
>
> private static String env(String key, String defaultValue) {
> String rc = System.getenv(key);
> if( rc== null )
> return defaultValue;
> return rc;
> }
>
> private static String arg(String []args, int index, String
> defaultValue)
> {
> if( index < args.length )
> return args[index];
> else
> return defaultValue;
> }
> }
>
> Thanks
>
>
>
> --
> View this message in context: http://activemq.2283324.n4.
> nabble.com/Network-of-Brokers-Connected-but-messages-are-
> not-received-on-second-broker-tp4720116p4720118.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>
