#activeMQ (6) - 서버 구현 및 메시지 송수신 테스트

테스트

- 다음과 같은 목표로 구현하였습니다.

- Broker1과  Broker2는 Broker3과 NetworkConnection되어있다. 

- Broker3은 Master 1,2는 slave 브로커로 구성한다.

시나리오

-> Broker3에 연결된 Producer가 메시지를 생산한다면, Broker1 과  2에 연결된 Consumer들 또한 메시지를 소비할 수 있어야 한다. 

<Server>

public class MQServer {


	public static void main(String[] args) {


		// TODO Auto-generated method stub


		//persistence & master slave 
		// DB 및 Persistence 설정
		KahaDBPersistenceAdapter persistenceAdapter = new KahaDBPersistenceAdapter();
	//	LevelDBPersistenceAdapter persistenceAdapter = new LevelDBPersistenceAdapter();
		persistenceAdapter.setDirectory(new File("ActiveMQ_Persistence","shared"));
		persistenceAdapter.setUseLock(false);
		
		
		
		// jmx managementcontext 생성
		ManagementContext managementContext = new ManagementContext();
		managementContext.setConnectorPath("/jmxrmi2");
		//managementContext.setMBeanServer();
		managementContext.setConnectorPort(1099);
		managementContext.setJmxDomainName("ActiveMQ Sever");
		managementContext.setUseMBeanServer(true);
		
		
		
		// 대형 메시징 패브릭의 대규모 확장 성을 제공하기 위해 일반적으로 많은 브로커를 네트워크에 연결하여 모든 논리적으로 연결된 클라이언트를
		// 원하는만큼 보유 할 수 있습니다. 필요에 따라 많은 메시지 브로커를 실행할 수 있습니다.
		
		try {
			//BrokerService broker = new BrokerService();
			BrokerService broker3 = BrokerFactory.createBroker("broker:()/master");
			BrokerService broker = BrokerFactory.createBroker("broker:()/slave");
			BrokerService broker2 = BrokerFactory.createBroker("broker:()/slave");


			
			//jmx
			broker.setManagementContext(managementContext);
			broker2.setManagementContext(managementContext);
			broker3.setManagementContext(managementContext);
			
			///advisory
			List<Object> entries = new ArrayList<>();
			PolicyEntry entry1 = new PolicyEntry();
			entry1.setAdvisoryForConsumed(true);
			entry1.setTopic("TEST");
		
			PolicyEntry entry2 = new PolicyEntry();
			entry2.setAdvisoryForConsumed(true);
			entry2.setQueue("TEST");
		
			entries.add(entry1);
			entries.add(entry2);
			
			PolicyMap policyMap = new PolicyMap();
			policyMap.setPolicyEntries(entries);
			
			// advisory Message setting
			broker.setAdvisorySupport(true);
			broker.setDestinationPolicy(policyMap);
		
			
			broker2.setAdvisorySupport(true);
			broker2.setDestinationPolicy(policyMap);
			
			broker3.setAdvisorySupport(true);
			broker3.setDestinationPolicy(policyMap);
			
			
			
			try {


				// broker.addConnector("tcp://localhost:61616");
				// java Management Extention


				broker.setBrokerName("Broker1");
				broker.setUseJmx(true); // check true or false
				broker.setPersistent(true);
				// broker.addConnector("tcp://localhost:61616");


				/// start() 메소드가 저장소 잠금 보류를 차단할 때 유용합니다 (예 : 슬레이브 시작).
				TransportConnector conn = new TransportConnector();
				conn.setUri(new URI("tcp://localhost:61616"));


				// failover:// 브로커 클러스터에 대한 클라이언트 연결을 업데이트
				// conn.setUpdateClusterClients(true);
				broker.setPersistenceAdapter(persistenceAdapter);
				broker.addConnector(conn);
				broker.start();
			
				
				
				broker2.setBrokerName("Broker2");
				broker2.setUseJmx(true); // check true or false
				broker2.setPersistent(true);
				broker2.setShutdownOnMasterFailure(true);
				//broker2.addConnector("tcp://172.10.11.20:61617");
			
				TransportConnector conn2 = new TransportConnector();
				conn2.setUri(new URI("tcp://localhost:61617"));
				broker2.setPersistenceAdapter(persistenceAdapter);
				broker2.addConnector(conn2);
				broker2.start();


				///////////////////////////////// Network Connect 사용


				// String networkConnectorURIs[] =
				// {"static:(tcp://localhost:61617)","static:(tcp://localhost:61616)"};


				/// 속성
				// property default description


				// initialReconnectDelay 1000 재접속을 시도하기까지 대기 할 시간 (ms) (useExponentialBackOff가
				// false 인 경우)
				// maxReconnectDelay 30000 다시 연결을 시도하기 전에 대기 할 시간 (ms)
				// useExponentialBackOff true 재 연결 시퀀스의 모든 실패에 대해 재 연결 사이의 시간을 증가시킵니다.
				// backOffMultiplier 2 대기 시간 증가 지수에 곱함 useExponentialBackOff true이면


				// 정적 발견
			/*	String networkConnectorURIs[] = {
						"static:(tcp://172.10.11.20:61617,tcp://172.10.11.20:61616)?maxReconnectDelay=5000&useExponentialBackOff=false" };
			*/
				/// static == masterslave
				
				String networkConnectorURIs[] = {
				"static:(tcp://localhost:61616,tcp://localhost:61617)?maxReconnectDelay=5000&useExponentialBackOff=false" };


				String transportConnectorURIs[] = { "tcp://localhost:61602" };


				broker3.setBrokerName("Broker3");
				broker3.setPersistent(true);
				broker3.setUseJmx(true);
			
				broker3.setPersistenceAdapter(persistenceAdapter);
				broker3.setTransportConnectorURIs(transportConnectorURIs);
				broker3.setNetworkConnectorURIs(networkConnectorURIs);
				
				//// persistenceAdapter share file system
				// broker3.deleteAllMessages();


				//


				// // 일반적으로 네트워크 브로커 시작은 순차적으로 시작함 . 비동기 시작을 위한 설정.
				broker3.setNetworkConnectorStartAsync(true);
				broker3.start();
				



			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		} catch (Exception e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
	}


}		

=> 서버를 실행하면 61602포트는 Broker3의 메시지 송수신 포트, 61616, 61617 포트는 각각  Broker1,2 와 커넥션 할 준비가 되어있습니다. 또한 서버 내 Broker3은 Network 연결된 것을 확인할 수 있습니다.

<Producer>

-> failover Protocol은 커넥션이 실패한다면 자동으로 다음 로컬주소를 발견하여 커넥션 시도를 합니다. 

		public class Text_Send {


	// URL of the JMS server. DEFAULT_BROKER_URL will just mean that JMS server is
	// on localhost
	//private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	
	///61613 에 실패
	private static String url = "failover:(tcp://localhost:61602)?randomize=false";
	
	// default broker URL is : tcp://localhost:61616"
	private static String subject = "TEST"; // Queue Name.You can create any/many queue names as per your
													// requirement.


	public static void main(String[] args) throws JMSException {
		// Getting JMS connection from the server and starting it
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
		Connection connection = connectionFactory.createConnection();


		connection.start();


		// Creating a non transactional session to send/receive JMS message.
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		


		
		// Destination represents here our queue 'JCG_QUEUE' on the JMS server.
		// The queue will be created automatically on the server.
		
		//Destination destination = session.createQueue(subject);
		Destination destination = session.createTopic(subject);
		
		// MessageProducer is used for sending messages to the queue.
		MessageProducer producer = session.createProducer(destination);


		
		
		for(int i=0; i<100; i++) {
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			// We will send a small text message saying 'Hello World!!!'
			//TextMessage message = session.createTextMessage("Hello !!!Send Message 1 Test");
			TextMessage message = session.createTextMessage("Hello !!!Send Message 1 Test");
			
			
			// Here we are sending our message!
			producer.send(message);
			System.out.println("JMS printing@@ '" + message.getText() + "'");
		}
		
	


		connection.close();
	}

<Consumer>

		


public class Receive2 {


	// URL of the JMS server
//	private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	
	private static String url ="failover:(tcp://localhost:61617,tcp://localhost:61616,tcp://localhost:61602)";
	
	// default broker URL is : tcp://localhost:61616"


	// Name of the queue we will receive messages from
	private static String subject = "TEST";


	public static void main(String[] args) throws JMSException {
		// Getting JMS connection from the server
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
		Connection connection = connectionFactory.createConnection();
		connection.start();


		// Creating session for seding messages
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);


		// Getting the queue 'JCG_QUEUE'
		//Destination destination = session.createQueue(subject);
		Destination destination = session.createTopic(subject);
		
		
		
		// MessageConsumer is used for receiving (consuming) messages
		MessageConsumer consumer = session.createConsumer(destination);
	
		while(true) {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			// Here we receive the message. 		/// block mode
			Message message = consumer.receive();


			// We will be using TestMessage in our example. MessageProducer sent us a
			// TextMessage
			// so we must cast to it to get access to its .getText() method.
			if (message instanceof TextMessage) {
				TextMessage textMessage = (TextMessage) message;
				System.out.println("Received message '" + textMessage.getText() + "'");
			}


			
		}
		
	}

=> 차례로 실행하면 61602 포트로 Broker3에 생산된 메시지를 , Broker2에 61616 포트로 연결된 Consumer가 메시지를 소비할 수 있습니다.

<결과>

간단하게 브로커를 직접 만들고 Clustering 기술을 활용하여 서버를 구현하고 테스트하는 실습하였습니다.

이외에도 jmx monitoring , HermesJMS 같은 third party tool과 연계, java 이외에 c, c++ 클라이언트와의 연결 advisory message, Object message, blob message 등의 특성을 활용할 수 있습니다.