Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
Ejemplos prácticos del uso de Java Message Service (JMS) con ActiveMQ
Los siguientes ejemplos muestran cómo puede trabajar con ActiveMQ mediante programación:
-
El código Java de OpenWire ejemplo se conecta a un intermediario, crea una cola y envía y recibe un mensaje. Para obtener un desglose y una explicación más detallados, consulte Connecting a Java application to your broker.
-
El ejemplo MQTT de código Java se conecta con un agente, crea un tema y publica y recibe un mensaje.
-
El ejemplo STOMP+WSS de código Java se conecta con un agente, crea una cola y publica y recibe un mensaje.
Requisitos previos
Habilitar atributos de VPC
Para asegurarse de poder acceder a su agente dentro de su VPC, debe habilitar los atributos enableDnsHostnames y enableDnsSupport de VPC. Para obtener más información, consulte el tema sobre soporte de DNS en su VPC en la Guía del usuario de Amazon VPC.
Habilitar las conexiones entrantes
Inicie sesión en la consola de Amazon MQ.
En la lista de corredores, elija el nombre de su corredor (por ejemplo, MyBroker).
-
En la MyBrokerpágina, en la sección Conexiones, anote las direcciones y los puertos de la URL de la consola web del bróker y de los protocolos a nivel de cable.
-
En la sección Details (Detalles), en Security and network (Seguridad y red), elija el nombre de su grupo de seguridad o
.
Aparecerá la página Security Groups (Grupos de seguridad) del panel de EC2.
-
En la lista de grupos de seguridad, elija el grupo de seguridad.
-
En la parte inferior de la página, elija Inbound (Entrante) y, a continuación, Edit (Editar).
-
En el cuadro de diálogo Edit inbound rules (Editar reglas entrantes), agregue una regla para cada URL o punto de enlace que desee que sea accesible públicamente (en el ejemplo a continuación se muestra cómo hacerlo para una consola web de agente).
-
Seleccione Add Rule (Agregar regla).
-
En Type (Tipo), seleccione Custom TCP (TCP personalizada).
-
Para Port Range (Rango de puertos), escriba el puerto de la consola web (8162).
-
Para Source (Fuente), deje seleccionada la opción Custom (Personalizada) y, luego, escriba la dirección IP del sistema que desea que pueda acceder a la consola web (por ejemplo, 192.0.2.1).
-
Seleccione Guardar.
El agente ahora puede aceptar las conexiones entrantes.
Agregar dependencias de Java
- OpenWire
-
Agregue los paquetes activemq-client.jar y activemq-pool.jar a la ruta de clases de Java. Los siguientes ejemplos muestran estas dependencias en el archivo pom.xml de un proyecto Maven.
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>5.15.16</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.15.16</version>
</dependency>
</dependencies>
Para obtener más información acerca de activemq-client.jar, consulte el tema sobre la configuración inicial en la documentación de Apache ActiveMQ.
- MQTT
-
Añada el paquete org.eclipse.paho.client.mqttv3.jar en la ruta de clases de Java. En el siguiente ejemplo se muestra esta dependencia en el archivo pom.xml de un proyecto Maven.
<dependencies>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.0</version>
</dependency>
</dependencies>
Para obtener más información sobre org.eclipse.paho.client.mqttv3.jar, consulte Cliente Java de Eclipse Paho.
- STOMP+WSS
-
Añada los paquetes siguientes a la ruta de clases de Java:
-
spring-messaging.jar
-
spring-websocket.jar
-
javax.websocket-api.jar
-
jetty-all.jar
-
slf4j-simple.jar
-
jackson-databind.jar
Los siguientes ejemplos muestran estas dependencias en el archivo pom.xml de un proyecto Maven.
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-messaging</artifactId>
<version>5.0.5.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-websocket</artifactId>
<version>5.0.5.RELEASE</version>
</dependency>
<dependency>
<groupId>javax.websocket</groupId>
<artifactId>javax.websocket-api</artifactId>
<version>1.1</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.aggregate</groupId>
<artifactId>jetty-all</artifactId>
<type>pom</type>
<version>9.3.3.v20150827</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.6.6</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.5.0</version>
</dependency>
</dependencies>
Para obtener más información, consulte el tema sobre la compatibilidad con STOMP en la documentación de la plataforma Spring.
AmazonMQExample.java
En el siguiente código de ejemplo, los productores y los consumidores se ejecutan en un único subproceso. Para sistemas de producción (o para probar la conmutación por error de instancias del agente), asegúrese de que sus productores y consumidores se ejecuten en anfitriones o subprocesos independientes.
- OpenWire
-
/*
* Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* https://aws.amazon.com/apache2.0
*
* or in the "license" file accompanying this file. This file 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.
*
*/
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.jms.pool.PooledConnectionFactory;
import javax.jms.*;
public class AmazonMQExample {
// Specify the connection parameters.
private final static String WIRE_LEVEL_ENDPOINT
= "ssl://b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9-1.mq.us-east-2.amazonaws.com:61617";
private final static String ACTIVE_MQ_USERNAME = "MyUsername123";
private final static String ACTIVE_MQ_PASSWORD = "MyPassword456";
public static void main(String[] args) throws JMSException {
final ActiveMQConnectionFactory connectionFactory =
createActiveMQConnectionFactory();
final PooledConnectionFactory pooledConnectionFactory =
createPooledConnectionFactory(connectionFactory);
sendMessage(pooledConnectionFactory);
receiveMessage(connectionFactory);
pooledConnectionFactory.stop();
}
private static void
sendMessage(PooledConnectionFactory pooledConnectionFactory) throws JMSException {
// Establish a connection for the producer.
final Connection producerConnection = pooledConnectionFactory
.createConnection();
producerConnection.start();
// Create a session.
final Session producerSession = producerConnection
.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Create a queue named "MyQueue".
final Destination producerDestination = producerSession
.createQueue("MyQueue");
// Create a producer from the session to the queue.
final MessageProducer producer = producerSession
.createProducer(producerDestination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// Create a message.
final String text = "Hello from Amazon MQ!";
final TextMessage producerMessage = producerSession
.createTextMessage(text);
// Send the message.
producer.send(producerMessage);
System.out.println("Message sent.");
// Clean up the producer.
producer.close();
producerSession.close();
producerConnection.close();
}
private static void
receiveMessage(ActiveMQConnectionFactory connectionFactory) throws JMSException {
// Establish a connection for the consumer.
// Note: Consumers should not use PooledConnectionFactory.
final Connection consumerConnection = connectionFactory.createConnection();
consumerConnection.start();
// Create a session.
final Session consumerSession = consumerConnection
.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Create a queue named "MyQueue".
final Destination consumerDestination = consumerSession
.createQueue("MyQueue");
// Create a message consumer from the session to the queue.
final MessageConsumer consumer = consumerSession
.createConsumer(consumerDestination);
// Begin to wait for messages.
final Message consumerMessage = consumer.receive(1000);
// Receive the message when it arrives.
final TextMessage consumerTextMessage = (TextMessage) consumerMessage;
System.out.println("Message received: " + consumerTextMessage.getText());
// Clean up the consumer.
consumer.close();
consumerSession.close();
consumerConnection.close();
}
private static PooledConnectionFactory
createPooledConnectionFactory(ActiveMQConnectionFactory connectionFactory) {
// Create a pooled connection factory.
final PooledConnectionFactory pooledConnectionFactory =
new PooledConnectionFactory();
pooledConnectionFactory.setConnectionFactory(connectionFactory);
pooledConnectionFactory.setMaxConnections(10);
return pooledConnectionFactory;
}
private static ActiveMQConnectionFactory createActiveMQConnectionFactory() {
// Create a connection factory.
final ActiveMQConnectionFactory connectionFactory =
new ActiveMQConnectionFactory(WIRE_LEVEL_ENDPOINT);
// Pass the sign-in credentials.
connectionFactory.setUserName(ACTIVE_MQ_USERNAME);
connectionFactory.setPassword(ACTIVE_MQ_PASSWORD);
return connectionFactory;
}
}
- MQTT
-
/*
* Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* https://aws.amazon.com/apache2.0
*
* or in the "license" file accompanying this file. This file 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.
*
*/
import org.eclipse.paho.client.mqttv3.*;
public class AmazonMQExampleMqtt implements MqttCallback {
// Specify the connection parameters.
private final static String WIRE_LEVEL_ENDPOINT =
"ssl://b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9-1.mq.us-east-2.amazonaws.com:8883";
private final static String ACTIVE_MQ_USERNAME = "MyUsername123";
private final static String ACTIVE_MQ_PASSWORD = "MyPassword456";
public static void main(String[] args) throws Exception {
new AmazonMQExampleMqtt().run();
}
private void run() throws MqttException, InterruptedException {
// Specify the topic name and the message text.
final String topic = "myTopic";
final String text = "Hello from Amazon MQ!";
// Create the MQTT client and specify the connection options.
final String clientId = "abc123";
final MqttClient client = new MqttClient(WIRE_LEVEL_ENDPOINT, clientId);
final MqttConnectOptions connOpts = new MqttConnectOptions();
// Pass the sign-in credentials.
connOpts.setUserName(ACTIVE_MQ_USERNAME);
connOpts.setPassword(ACTIVE_MQ_PASSWORD.toCharArray());
// Create a session and subscribe to a topic filter.
client.connect(connOpts);
client.setCallback(this);
client.subscribe("+");
// Create a message.
final MqttMessage message = new MqttMessage(text.getBytes());
// Publish the message to a topic.
client.publish(topic, message);
System.out.println("Published message.");
// Wait for the message to be received.
Thread.sleep(3000L);
// Clean up the connection.
client.disconnect();
}
@Override
public void connectionLost(Throwable cause) {
System.out.println("Lost connection.");
}
@Override
public void messageArrived(String topic, MqttMessage message) throws MqttException {
System.out.println("Received message from topic " + topic + ": " + message);
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("Delivered message.");
}
}
- STOMP+WSS
-
/*
* Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* https://aws.amazon.com/apache2.0
*
* or in the "license" file accompanying this file. This file 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.
*
*/
import org.springframework.messaging.converter.StringMessageConverter;
import org.springframework.messaging.simp.stomp.*;
import org.springframework.web.socket.WebSocketHttpHeaders;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.messaging.WebSocketStompClient;
import java.lang.reflect.Type;
public class AmazonMQExampleStompWss {
// Specify the connection parameters.
private final static String DESTINATION = "/queue";
private final static String WIRE_LEVEL_ENDPOINT =
"wss://b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9-1.mq.us-east-2.amazonaws.com:61619";
private final static String ACTIVE_MQ_USERNAME = "MyUsername123";
private final static String ACTIVE_MQ_PASSWORD = "MyPassword456";
public static void main(String[] args) throws Exception {
final AmazonMQExampleStompWss example = new AmazonMQExampleStompWss();
final StompSession stompSession = example.connect();
System.out.println("Subscribed to a destination using session.");
example.subscribeToDestination(stompSession);
System.out.println("Sent message to session.");
example.sendMessage(stompSession);
Thread.sleep(60000);
}
private StompSession connect() throws Exception {
// Create a client.
final WebSocketClient client = new StandardWebSocketClient();
final WebSocketStompClient stompClient = new WebSocketStompClient(client);
stompClient.setMessageConverter(new StringMessageConverter());
final WebSocketHttpHeaders headers = new WebSocketHttpHeaders();
// Create headers with authentication parameters.
final StompHeaders head = new StompHeaders();
head.add(StompHeaders.LOGIN, ACTIVE_MQ_USERNAME);
head.add(StompHeaders.PASSCODE, ACTIVE_MQ_PASSWORD);
final StompSessionHandler sessionHandler = new MySessionHandler();
// Create a connection.
return stompClient.connect(WIRE_LEVEL_ENDPOINT, headers, head,
sessionHandler).get();
}
private void subscribeToDestination(final StompSession stompSession) {
stompSession.subscribe(DESTINATION, new MyFrameHandler());
}
private void sendMessage(final StompSession stompSession) {
stompSession.send(DESTINATION, "Hello from Amazon MQ!".getBytes());
}
private static class MySessionHandler extends StompSessionHandlerAdapter {
public void afterConnected(final StompSession stompSession,
final StompHeaders stompHeaders) {
System.out.println("Connected to broker.");
}
}
private static class MyFrameHandler implements StompFrameHandler {
public Type getPayloadType(final StompHeaders headers) {
return String.class;
}
public void handleFrame(final StompHeaders stompHeaders,
final Object message) {
System.out.print("Received message from topic: " + message);
}
}
}
