QPID通讯代码01

• TextMessage
• ListMessage
• MapMesage
• StreamMessage

1、MqConsumerText.java

package com.neohope.qpid.test;

import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.util.Properties;

public class MqConsumerText {

    public static void main(String[] args)
    {
        try
        {
            Properties properties = new Properties();
            properties.setProperty("java.naming.factory.initial", "org.apache.qpid.jndi.PropertiesFileInitialContextFactory");
            properties.setProperty("connectionfactory.qpidConnectionfactory","amqp://guest:guest@clientid/?brokerlist='tcp://localhost:5672'");
            properties.setProperty("destination.topicExchange","amq.topic");

            Context context = new InitialContext(properties);

            ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup("qpidConnectionfactory");
            Connection connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = (Destination) context.lookup("topicExchange");

            MessageConsumer messageConsumer = session.createConsumer(destination);
            TextMessage message = (TextMessage)messageConsumer.receive();
            System.out.println(message.getText());

            connection.close();
            context.close();
        }
        catch (Exception ex)
        {
            ex.printStackTrace();
        }
    }
}

2、MqProducerText.java

package com.neohope.qpid.test;

import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.io.InputStream;
import java.util.Properties;

public class MqProducerText {
    public static void main(String[] args)
    {
        try
        {
            Properties properties = new Properties();
            properties.setProperty("java.naming.factory.initial", "org.apache.qpid.jndi.PropertiesFileInitialContextFactory");
            properties.setProperty("connectionfactory.qpidConnectionfactory","amqp://guest:guest@clientid/?brokerlist='tcp://localhost:5672'");
            properties.setProperty("destination.topicExchange","amq.topic");

            Context context = new InitialContext(properties);

            ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup("qpidConnectionfactory");
            Connection connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = (Destination) context.lookup("topicExchange");

            MessageProducer messageProducer = session.createProducer(destination);
            TextMessage message = session.createTextMessage("Hello Qpid!");
            messageProducer.send(message);

            connection.close();
            context.close();
        }
        catch (Exception ex)
        {
            ex.printStackTrace();
        }
    }
}

3、测试

#开启qpid-broker
#运行MqConsumerText.Main
#运行MqProducerText.Main