使用RabbitMQ Java Client
Publishing Messages 连接到Broker 获取Channel 声明一个Exchange 创建一个Message 发布这个Message 关闭Channel 关闭Connection ConnectionFactory connectionFactory = new ConnectionFactory(); // AMQP URI amqp://userName:password@hostName:portNumber/virtualHost" connectionFactory.setUri("amqp://test:[email protected]:5672/demo"); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("first-exchange", "direct", true); byte[] msgBody = "Hello, World!".getBytes("UTF-8"); channel.basicPublish("first-exchange", "helloRoutingKey", null, msgBody); channel.close(); connection.close(); Receiving Messages by Subscription public static void main(String[] args) throws Exception { ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setUri("amqp://test:[email protected]:5672/demo"); Connection connection = connectionFactory.newConnection(); final Channel channel = connection.createChannel(); channel.exchangeDeclare("first-exchange", "direct", true); channel.queueDeclare("hello-queue", true, false, false, null); channel.queueBind("hello-queue", "first-exchange", "helloRoutingKey"); channel.basicConsume("hello-queue", new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "UTF-8"); System.out.println(msg); channel.basicAck(envelope.getDeliveryTag(), false); } }); Scanner scanner = new Scanner(System.in); scanner.nextLine(); scanner.close(); channel.close(); connection.close(); }