当前位置:我的异常网» 软件架构设计 » （java）简略实现原生RabbitMQ中的广播订阅（fanout

（java）简略实现原生RabbitMQ中的广播订阅（fanout）模式

www.myexceptions.net 网友分享于：2018-06-06 浏览：0次

（java）简单实现原生RabbitMQ中的广播订阅（fanout）模式

fanout 发送端

import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
  
public class MyFanoutSender {  
      
     
    private static final String MESSAGE = "my name is";  
      
    public static void main(String[] args){  
        Connection conn = null;  
        Channel channel = null;  
        try {  
            //初始化连接，主机，端口，用户名，密码可以自己定义  
            ConnectionFactory factory = new ConnectionFactory();  
            factory.setHost(HOST_PARAMETER.LOCAL_HOST);  
            factory.setPort(HOST_PARAMETER.LOCAL_PORT);  
            factory.setUsername(HOST_PARAMETER.LOCAL_USER_NAME);  
            factory.setPassword(HOST_PARAMETER.LOCAL_PASSWORD);  
            //创建连接  
            conn = factory.newConnection();  
            //创建通道  
            channel = conn.createChannel();  
            //定义为fanout类型的交换机  
            channel.exchangeDeclare(HOST_PARAMETER.EXCHANGE_NAME, "fanout");  
            //发送，指定routingkey为""  
            channel.basicPublish(HOST_PARAMETER.EXCHANGE_NAME, "", null, MESSAGE.getBytes());  
            System.out.println("I send a fanout massage!");  
        } catch (Exception e) {  
            e.printStackTrace();  
        } finally{  
            try {  
                if(channel != null){  
                    channel.close();  
                }  
                if(conn != null){  
                    conn.close();  
                }  
            } catch (Exception e) {  
                e.printStackTrace();  
            }  
        }  
    }  
}  

fanout订阅者（接收端），可以自己多定义几个，才能看出效果

import java.io.IOException;  
  
import com.rabbitmq.client.AMQP.BasicProperties;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
import com.rabbitmq.client.Consumer;  
import com.rabbitmq.client.DefaultConsumer;  
import com.rabbitmq.client.Envelope;  
  
public class MyFirstFanoutReceiver {  
      
  public static void main(String[] args){  
      Connection conn = null;  
      Channel channel = null;  
      try {  
        //初始化连接  
        ConnectionFactory factory = new ConnectionFactory();  
        factory.setHost(HOST_PARAMETER.LOCAL_HOST);  
        factory.setPort(HOST_PARAMETER.LOCAL_PORT);  
        factory.setUsername(HOST_PARAMETER.LOCAL_USER_NAME);  
        factory.setPassword(HOST_PARAMETER.LOCAL_PASSWORD);  
        //创建连接  
        conn = factory.newConnection();  
        //创建通道  
        channel = conn.createChannel();  
        //声明交换机类型  
        channel.exchangeDeclare(HOST_PARAMETER.EXCHANGE_NAME, "fanout");  
        //声明默认的队列  
        String queue = channel.queueDeclare().getQueue();  
        //将队列与交换机绑定，最后一个参数为routingKey,与发送者指定的一样""  
        channel.queueBind(queue, HOST_PARAMETER.EXCHANGE_NAME, "");  
        //消费者  
        Consumer consumer = new DefaultConsumer(channel){  
            @Override  
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)  
                    throws IOException {  
                System.out.println(new String(body,"utf-8")+" Tom");  
            }  
              
        };  
        channel.basicConsume(queue, true, consumer);  
        System.out.println("i am the first fanout receiver!");  
      }catch (Exception e) {  
            e.printStackTrace();  
      }   
  }  
}  

《使用HTML，CSS，JavaScript开发Android应用程序》英文PDF文字版资源

软件架构设计-热门软件架构设计-最新软件架构设计-其它

（java）简略实现原生RabbitMQ中的广播订阅（fanout）模式

文章评论