|
ActiveMQ example source code file (AMQDeadlockTestW4Brokers.java)
The ActiveMQ AMQDeadlockTestW4Brokers.java source code/** * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package org.apache.activemq.usecases; import java.net.URI; import java.util.ArrayList; import java.util.List; import java.util.Random; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import javax.jms.BytesMessage; import javax.jms.ConnectionFactory; import javax.jms.DeliveryMode; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.Session; import junit.framework.TestCase; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.test.*; import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.TransportConnector; import org.apache.activemq.broker.region.policy.PolicyEntry; import org.apache.activemq.broker.region.policy.PolicyMap; import org.apache.activemq.network.DiscoveryNetworkConnector; import org.apache.activemq.network.NetworkConnector; import org.apache.activemq.pool.PooledConnectionFactory; import org.apache.activemq.usage.SystemUsage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import org.springframework.jms.listener.DefaultMessageListenerContainer; public class AMQDeadlockTestW4Brokers extends org.apache.activemq.test.TestSupport { private static final transient Logger LOG = LoggerFactory.getLogger(AMQDeadlockTestW4Brokers.class); private static final String BROKER_URL1 = "tcp://localhost:61616"; private static final String BROKER_URL2 = "tcp://localhost:61617"; private static final String BROKER_URL3 = "tcp://localhost:61618"; private static final String BROKER_URL4 = "tcp://localhost:61619"; private static final String URL1 = "tcp://localhost:61616?wireFormat.cacheEnabled=false" + "&wireFormat.tightEncodingEnabled=false&wireFormat.maxInactivityDuration=30000&wireFormat.tcpNoDelayEnabled=false"; private static final String URL2 = "tcp://localhost:61617?wireFormat.cacheEnabled=false" + "&wireFormat.tightEncodingEnabled=false&wireFormat.maxInactivityDuration=30000&wireFormat.tcpNoDelayEnabled=false"; private static final String URL3 = "tcp://localhost:61618?wireFormat.cacheEnabled=false" + "&wireFormat.tightEncodingEnabled=false&wireFormat.maxInactivityDuration=30000&wireFormat.tcpNoDelayEnabled=false"; private static final String URL4 = "tcp://localhost:61619?wireFormat.cacheEnabled=false" + "&wireFormat.tightEncodingEnabled=false&wireFormat.maxInactivityDuration=30000&wireFormat.tcpNoDelayEnabled=false"; private static final String QUEUE1_NAME = "test.queue.1"; private static final int MAX_CONSUMERS = 5; private static final int NUM_MESSAGE_TO_SEND = 10000; private static final CountDownLatch LATCH = new CountDownLatch(MAX_CONSUMERS * NUM_MESSAGE_TO_SEND); @Override public void setUp() throws Exception { } @Override public void tearDown() throws Exception { } public void test4BrokerWithOutLingo() throws Exception { BrokerService brokerService1 = null; BrokerService brokerService2 = null; BrokerService brokerService3 = null; BrokerService brokerService4 = null; ActiveMQConnectionFactory acf1 = null; ActiveMQConnectionFactory acf2 = null; PooledConnectionFactory pcf1 = null; PooledConnectionFactory pcf2 = null; ActiveMQConnectionFactory acf3 = null; ActiveMQConnectionFactory acf4 = null; PooledConnectionFactory pcf3 = null; PooledConnectionFactory pcf4 = null; DefaultMessageListenerContainer container1 = null; try { // Test with and without queue limits. brokerService1 = createBrokerService("broker1", BROKER_URL1, BROKER_URL2, BROKER_URL3, BROKER_URL4, 0 /* 10000000 */); brokerService1.start(); brokerService2 = createBrokerService("broker2", BROKER_URL2, BROKER_URL1, BROKER_URL3, BROKER_URL4, 0/* 40000000 */); brokerService2.start(); brokerService3 = createBrokerService("broker3", BROKER_URL3, BROKER_URL2, BROKER_URL1, BROKER_URL4, 0/* 10000000 */); brokerService3.start(); brokerService4 = createBrokerService("broker4", BROKER_URL4, BROKER_URL1, BROKER_URL3, BROKER_URL2, 0/* 10000000 */); brokerService4.start(); final String failover1 = "failover:(" + URL1 + ")?initialReconnectDelay=10&maxReconnectDelay=30000&useExponentialBackOff=true&backOffMultiplier=2&maxReconnectAttempts=0&randomize=false"; final String failover2 = "failover:(" + URL2 + ")?initialReconnectDelay=10&maxReconnectDelay=30000&useExponentialBackOff=true&backOffMultiplier=2&maxReconnectAttempts=0&randomize=false"; final String failover3 = "failover:(" + URL3 + ")?initialReconnectDelay=10&maxReconnectDelay=30000&useExponentialBackOff=true&backOffMultiplier=2&maxReconnectAttempts=0&randomize=false"; final String failover4 = "failover:(" + URL4 + ")?initialReconnectDelay=10&maxReconnectDelay=30000&useExponentialBackOff=true&backOffMultiplier=2&maxReconnectAttempts=0&randomize=false"; acf1 = createConnectionFactory(failover1); acf2 = createConnectionFactory(failover2); acf3 = createConnectionFactory(failover3); acf4 = createConnectionFactory(failover4); pcf1 = new PooledConnectionFactory(acf1); pcf2 = new PooledConnectionFactory(acf2); pcf3 = new PooledConnectionFactory(acf3); pcf4 = new PooledConnectionFactory(acf4); container1 = createDefaultMessageListenerContainer(acf2, new TestMessageListener1(0), QUEUE1_NAME); container1.afterPropertiesSet(); final PooledProducerTask[] task = new PooledProducerTask[4]; task[0] = new PooledProducerTask(pcf1, QUEUE1_NAME, "producer1"); task[1] = new PooledProducerTask(pcf2, QUEUE1_NAME, "producer2"); task[2] = new PooledProducerTask(pcf3, QUEUE1_NAME, "producer3"); task[3] = new PooledProducerTask(pcf4, QUEUE1_NAME, "producer4"); final ExecutorService executor = Executors.newCachedThreadPool(); for (int i = 0; i < 4; i++) { executor.submit(task[i]); } LATCH.await(15, TimeUnit.SECONDS); assertTrue(LATCH.getCount() == MAX_CONSUMERS * NUM_MESSAGE_TO_SEND); } catch (Exception e) { e.printStackTrace(); } finally { container1.stop(); container1.destroy(); container1 = null; brokerService1.stop(); brokerService1 = null; brokerService2.stop(); brokerService2 = null; brokerService3.stop(); brokerService3 = null; brokerService4.stop(); brokerService4 = null; } } private BrokerService createBrokerService(final String brokerName, final String uri1, final String uri2, final String uri3, final String uri4, final int queueLimit) throws Exception { final BrokerService brokerService = new BrokerService(); brokerService.setBrokerName(brokerName); brokerService.setPersistent(false); brokerService.setUseJmx(true); final SystemUsage memoryManager = new SystemUsage(); memoryManager.getMemoryUsage().setLimit(100000000); brokerService.setSystemUsage(memoryManager); final List<PolicyEntry> policyEntries = new ArrayList Other ActiveMQ examples (source code examples)Here is a short list of links related to this ActiveMQ AMQDeadlockTestW4Brokers.java source code file: |
... this post is sponsored by my books ... | |
#1 New Release! |
FP Best Seller |
Copyright 1998-2021 Alvin Alexander, alvinalexander.com
All Rights Reserved.
A percentage of advertising revenue from
pages under the /java/jwarehouse
URI on this website is
paid back to open source projects.