devbox
人工智能uu
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ管理端代码（开箱即用）_createchannel(false)

作者：人工智能uu | 2024-08-13 20:58:21

赞

踩

createchannel(false)

骨架图：

以下代码开箱即用，直接复制即可（注意MQ配置读取的是个人配置中心的MQ配置信息）

 

依次介绍：

 一、MQ配置：RabbitConfig

通过@Bean注入了Spring  IOC容器中

  1. @Configuration
  2. public class RabbitConfig {
  3.  
  4.   @Value("${mqRabbitHost}")
  5.   private String addresses;//设置连接地址
  6.  
  7.   @Value("${mqRabbitUserName}")
  8.   public String username;//设置用户名称
  9.  
  10.   @Value("${mqRabbitPassword}")
  11.   public String password;//设置用户密码
  12.  
  13. //权限控制的基本单位，一个VirtualHost里面有若干Exchange和MessageQueue，以及指定被哪些user或者团队使用
  14.   @Value("${mqRabbitVirtualHost}")
  15.   private String virtualHost;//设置VirtualHost地址
  16.  
  17.   @Value("${mqRabbitPublisher-confirms}")
  18.   private boolean publisherConfirms;//发行者确认
  19.  
  20.   @Value("${mqRabbitApi-url}")
  21.   public String URL;//
  22.  
  23.   @Bean
  24.   public ConnectionFactory connectionFactory() {
  25. //创建连接工厂
  26.     CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
  27.     connectionFactory.setAddresses(addresses);
  28.     connectionFactory.setUsername(username);
  29.     connectionFactory.setPassword(password);
  30.     connectionFactory.setVirtualHost(virtualHost);
  31.     /** 如果要进行消息回调，则这里必须要设置为true */
  32.     connectionFactory.setPublisherConfirms(publisherConfirms);
  33.     return connectionFactory;
  34.   }
  35.  
  36.  
  37.   /**
  38.    * 因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置
  39.    * 
  40.    * @return
  41.    */
  42.   @Bean
  43.   @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  44.   public RabbitTemplate rabbitTemplatenew() {
  45.     RabbitTemplate template = new RabbitTemplate(connectionFactory());
  46.     return template;
  47.   }
  48.  
  49.   public String getUsername() {
  50.     return username;
  51.   }
  52.  
  53.   public String getPassword() {
  54.     return password;
  55.   }
  56.  
  57.  
  58. }

二、entity

（1）Binding：绑定

  1. public class Binding {
  2.  
  3.   /**
  4.    * 绑定源
  5.    */
  6.   private String source;
  7.   /**
  8.    * 虚拟主机
  9.    */
  10.   private String vhost;
  11.   /**
  12.    * 绑定目标
  13.    */
  14.   private String destination;
  15.   /**
  16.    * 绑定的目标的类型
  17.    */
  18.   private String destination_type;
  19.   /**
  20.    * 路由键
  21.    */
  22.   private String routing_key;
  23.   /**
  24.    * 属性键
  25.    */
  26.   private String properties_key;
  27.  
  28.   get  and set...
  29. }

（2）Exchange：交换机

  1. public class Exchange {
  2.  
  3.   /**
  4.    * 是否持久化
  5.    */
  6.   private Boolean durable = true;
  7.   /**
  8.    * 交换机名称
  9.    */
  10.   private String name;
  11.   /**
  12.    * 原来交换机名称
  13.    */
  14.   private String oldName;
  15.   /**
  16.    * 虚拟主机
  17.    */
  18.   private String vhost;
  19.   /**
  20.    * 交换机类型
  21.    */
  22.   private String type = "fanout";
  23.   /**
  24.    * 是否只能交换机内部使用
  25.    */
  26.   private String internal;
  27.   /**
  28.    * 是否自动删除
  29.    */
  30.   private Boolean auto_delete = false;
  31.   
  32.   get and set...
  33. }

（3）Queue：队列

  1. public class Queue {
  2.   /**
  3.    * 队列运行状态
  4.    */
  5.   private String state;
  6.   /**
  7.    * 队列名称
  8.    */
  9.   private String name;
  10.   /**
  11.    * 原来队列名称
  12.    */
  13.   private String oldName;
  14.   /**
  15.    * 虚拟主机
  16.    */
  17.   private String vhost;
  18.   /**
  19.    * 是否持久化
  20.    */
  21.   private Boolean durable = true;
  22.   /**
  23.    * 是否自动删除
  24.    */
  25.   private Boolean auto_delete = false;
  26.   /**
  27.    * 是否排外
  28.    */
  29.   private Boolean exclusive = false;
  30.   /**
  31.    * 对应的节点
  32.    */
  33.   private String node;
  34.   
  35.   get and set..
  36. }

三、RabbitMqManageService

  1. @Transactional()
  2. public interface RabbitMqManageService {
  3.   /**
  4.    * 新增MQ队列
  5.    */
  6.   ResponseResult queueDeclare(Queue queue);
  7.   /**
  8.    * 修改MQ队列名称
  9.    */
  10.   ResponseResult queueModify(Queue queue);
  11.   /**
  12.    * 删除MQ队列
  13.    */
  14.   ResponseResult queueDelete(Queue queue);
  15.   /**
  16.    * 新增MQ交换机
  17.    */
  18.   ResponseResult exchangeDeclare(Exchange exchange);
  19.   /**
  20.    * 修改MQ交换机名称
  21.    */
  22.   ResponseResult exchangeModify(Exchange exchange);
  23.   /**
  24.    * 删除MQ交换机
  25.    */
  26.   ResponseResult exchangeDelete(Exchange exchange);
  27.   /**
  28.    * MQ交换机绑定队列
  29.    */
  30.   ResponseResult binding(Binding binding);
  31.   /**
  32.    * 交换机解绑队列
  33.    */
  34.   ResponseResult unbinding(Binding binding);
  35.  
  36. }

四：RabbitMqManageServiceImpl

  1. @Service
  2. public class RabbitMqManageServiceImpl implements RabbitMqManageService {
  3.  
  4.   @Autowired
  5.   private ConnectionFactory connectionFactory;
  6.  
  7.   @Autowired
  8.   private RabbitConfig config;
  9.  
  10.   @Override
  11.   public ResponseResult queueDeclare(Queue queue) {
  12.     // 参数 ：
  13.     // 队列名称，是否持久化，是否排外，是否自动删除，参数
  14.     // 队列名以 mia.服务名称.custom 命名
  15.     String name = queue.getName();
  16.     // boolean begin = name.startsWith("mia.");
  17.     // boolean end = name.endsWith(".custom");
  18.  
  19.     ResponseResult result = new ResponseResult();
  20.  
  21.     // if (!begin || !end) {
  22.     // throw new BusinessAccessException("队列命名方式不对");
  23.     // }
  24.     try {
  25.       AMQP.Queue.DeclareOk queueResult =
  26.           connectionFactory.createConnection().createChannel(false).queueDeclare(name,
  27.               queue.getDurable(), queue.getExclusive(), queue.getAuto_delete(), null);
  28.       String resultName = queueResult.getQueue();
  29.       if (resultName != null && resultName.equals(name)) {
  30.         result.setCode("200");
  31.         result.setMsg("success");
  32.       }
  33.     } catch (AmqpException e) {
  34.       e.printStackTrace();
  35.     } catch (IOException e) {
  36.       e.printStackTrace();
  37.     }
  38.     return result;
  39.   }
  40.  
  41.   @Override
  42.   public ResponseResult queueModify(Queue queue) {
  43.     String newName = queue.getName();
  44.     String oldName = queue.getOldName();
  45.     if (newName == null || oldName == null) {
  46.       throw new BusinessAccessException("请输入正确的参数");
  47.     }
  48.     queue.setName(oldName);
  49.     ResponseResult result = this.queueDelete(queue);
  50.     if ("200".equals(result.getCode())) {
  51.       queue.setName(newName);
  52.       result = this.queueDeclare(queue);
  53.       if ("200".equals(result.getCode())) {
  54.         return result;
  55.       } else {
  56.         throw new BusinessAccessException("修改队列失败，请直接新建队列");
  57.       }
  58.     }
  59.     return result;
  60.   }
  61.  
  62.   @Override
  63.   public ResponseResult queueDelete(Queue queue) {
  64.     // 参数 ：
  65.     // 队列名称，是否持久化，是否排外，是否自动删除，参数
  66.     // 队列名以 mia.服务名称.custom 命名
  67.     String name = queue.getName();
  68.     // boolean begin = name.startsWith("mia.");
  69.     // boolean end = name.endsWith(".custom");
  70.  
  71.     ResponseResult result = new ResponseResult();
  72.  
  73.     Queue queryResult = this.getQueue(name);
  74.     if (queryResult.getName() == null) {
  75.       result.setMsg("not found queue " + name);
  76.       return result;
  77.     }
  78.  
  79.     // if (!begin || !end) {
  80.     // throw new BusinessAccessException("只能删除自定义创建队列");
  81.     // }
  82.     try {
  83.       AMQP.Queue.DeleteOk queueResult =
  84.           connectionFactory.createConnection().createChannel(false).queueDelete(name);
  85.       if (queueResult.getMessageCount() >= 0) {
  86.         result.setCode("200");
  87.         result.setMsg("success");
  88.       }
  89.  
  90.     } catch (AmqpException e) {
  91.       e.printStackTrace();
  92.     } catch (IOException e) {
  93.       e.printStackTrace();
  94.     }
  95.     return result;
  96.   }
  97.  
  98.   /**
  99.    * 根据名称查询某个队列
  100.    * 
  101.    * @param name
  102.    * @return
  103.    */
  104.   public Queue getQueue(String name) {
  105.     String vHost = "/";
  106.     String url = config.URL + "queues/" + HttpUtil.getURLEncoderString(vHost) + "/" + name;
  107.     try {
  108.       String content = HttpUtil.get(url);
  109.       Queue result = JSON.parseObject(content, Queue.class);
  110.       return result;
  111.     } catch (Exception e) {
  112.       e.printStackTrace();
  113.     }
  114.     return null;
  115.   }
  116.  
  117.   /**
  118.    * 新增MQ交换机
  119.    * 
  120.    * @param exchange
  121.    */
  122.   @Override
  123.   public ResponseResult exchangeDeclare(Exchange exchange) {
  124.     // 参数 ：
  125.     // 交换机名称，类型，是否持久化，是否自动删除，参数
  126.     // 交换机名以 mia.业务名称.custom 命名
  127.  
  128.     String name = exchange.getName();
  129.  
  130.     // boolean begin = name.startsWith("mia.");
  131.     // boolean end = name.endsWith(".custom");
  132.  
  133.     ResponseResult result = new ResponseResult();
  134.  
  135.     // if (!begin || !end) {
  136.     // throw new BusinessAccessException("交换机命名方式不对");
  137.     // }
  138.  
  139.     String type = exchange.getType();
  140.     if (!"direct".equals(type) && !"fanout".equals(type) && !"topic".equals(type)
  141.         && !"headers".equals(type)) {
  142.       throw new BusinessAccessException("交换机的类型不对");
  143.     }
  144.  
  145.     try {
  146.       AMQP.Exchange.DeclareOk exchangeResult =
  147.           connectionFactory.createConnection().createChannel(false).exchangeDeclare(name, type,
  148.               exchange.getDurable(), exchange.getAuto_delete(), null);
  149.       if (exchangeResult != null) {
  150.         result.setCode("200");
  151.         result.setMsg("success");
  152.       }
  153.     } catch (AmqpException e) {
  154.       e.printStackTrace();
  155.     } catch (IOException e) {
  156.       e.printStackTrace();
  157.     }
  158.     return result;
  159.   }
  160.  
  161.   @Override
  162.   public ResponseResult exchangeModify(Exchange exchange) {
  163.     String newName = exchange.getName();
  164.     String oldName = exchange.getOldName();
  165.     if (newName == null || oldName == null) {
  166.       throw new BusinessAccessException("请输入正确的参数");
  167.     }
  168.     exchange.setName(oldName);
  169.     ResponseResult result = this.exchangeDelete(exchange);
  170.     if ("200".equals(result.getCode())) {
  171.       exchange.setName(newName);
  172.       result = this.exchangeDeclare(exchange);
  173.       if ("200".equals(result.getCode())) {
  174.         return result;
  175.       } else {
  176.         throw new BusinessAccessException("修改交换机失败，请直接新建交换机");
  177.       }
  178.     }
  179.     return result;
  180.   }
  181.  
  182.   @Override
  183.   public ResponseResult exchangeDelete(Exchange exchange) {
  184.     // 参数 ：
  185.     // 交换机名称，类型，是否持久化，是否自动删除，参数
  186.     // 交换机名以 mia.业务名称.custom 命名
  187.  
  188.     String name = exchange.getName();
  189.     // boolean begin = name.startsWith("mia.");
  190.     // boolean end = name.endsWith(".custom");
  191.  
  192.     ResponseResult result = new ResponseResult();
  193.  
  194.     Exchange queryResult = this.getExchange(name);
  195.     if (queryResult.getName() == null) {
  196.       result.setMsg("not found exchange " + name);
  197.       return result;
  198.     }
  199.  
  200.     if (queryResult.getName() == null) {
  201.       result.setMsg("not found queue " + name);
  202.       return result;
  203.     }
  204.  
  205.  
  206.     // if (!begin || !end) {
  207.     // throw new BusinessAccessException("只能删除自定义创建交换机");
  208.     // }
  209.     try {
  210.       AMQP.Exchange.DeleteOk exchangeResult =
  211.           connectionFactory.createConnection().createChannel(false).exchangeDelete(name);
  212.       if (exchangeResult != null) {
  213.         result.setCode("200");
  214.         result.setMsg("success");
  215.       }
  216.     } catch (AmqpException e) {
  217.       e.printStackTrace();
  218.     } catch (IOException e) {
  219.       e.printStackTrace();
  220.     }
  221.     return result;
  222.   }
  223.  
  224.   /**
  225.    * 根据名称查询某个交换机
  226.    * 
  227.    * @param name
  228.    * @return
  229.    */
  230.   public Exchange getExchange(String name) {
  231.  
  232.     String vHost = "/";
  233.  
  234.     String url = config.URL + "exchanges/" + HttpUtil.getURLEncoderString(vHost) + "/" + name;
  235.  
  236.     try {
  237.       String content = HttpUtil.get(url);
  238.       Exchange result = JSON.parseObject(content, Exchange.class);
  239.       return result;
  240.     } catch (Exception e) {
  241.       e.printStackTrace();
  242.     }
  243.     return null;
  244.   }
  245.  
  246.   @Override
  247.   public ResponseResult binding(Binding binding) {
  248.     // 参数 ：
  249.     // 队列名称，交换机名称，路由键值，参数
  250.  
  251.     String queueName = binding.getDestination();
  252.     String exchangeName = binding.getSource();
  253.     String key = binding.getRouting_key();
  254.  
  255.     ResponseResult result = new ResponseResult();
  256.     Queue queue = this.getQueue(queueName);
  257.     Exchange exchange = this.getExchange(exchangeName);
  258.     if (queue.getName() == null || exchange.getName() == null) {
  259.       result.setMsg("not found queue or exchange");
  260.       return result;
  261.     }
  262.  
  263.     if (queueName == null || exchangeName == null || key == null) {
  264.       throw new BusinessAccessException("关键参数不能为空");
  265.     }
  266.  
  267.     try {
  268.       AMQP.Queue.BindOk bindOk = connectionFactory.createConnection().createChannel(false)
  269.           .queueBind(queueName, exchangeName, key, null);
  270.       if (bindOk != null) {
  271.         result.setCode("200");
  272.         result.setMsg("success");
  273.       }
  274.     } catch (AmqpException e) {
  275.       e.printStackTrace();
  276.     } catch (IOException e) {
  277.       e.printStackTrace();
  278.     }
  279.     return result;
  280.   }
  281.  
  282.   @Override
  283.   public ResponseResult unbinding(Binding binding) {
  284.     // 参数 ：
  285.     // 队列名称，交换机名称，路由键值，参数
  286.  
  287.     String queueName = binding.getDestination();
  288.  
  289.     String exchangeName = binding.getSource();
  290.  
  291.     String key = binding.getRouting_key();
  292.  
  293.     ResponseResult result = new ResponseResult();
  294.  
  295.     Queue queue = this.getQueue(queueName);
  296.  
  297.     Exchange exchange = this.getExchange(exchangeName);
  298.     if (queue.getName() == null || exchange.getName() == null) {
  299.       result.setMsg("not found queue or exchange");
  300.       return result;
  301.     }
  302.  
  303.     if (queueName == null || exchangeName == null || key == null) {
  304.       throw new BusinessAccessException("关键参数不能为空");
  305.     }
  306.  
  307.     try {
  308.       AMQP.Queue.UnbindOk unbindOk = connectionFactory.createConnection().createChannel(false)
  309.           .queueUnbind(queueName, exchangeName, key, null);
  310.       if (unbindOk != null) {
  311.         result.setCode("200");
  312.         result.setMsg("success");
  313.       }
  314.  
  315.     } catch (AmqpException e) {
  316.       e.printStackTrace();
  317.     } catch (IOException e) {
  318.       e.printStackTrace();
  319.     }
  320.     return result;
  321.   }
  322.  
  323. }

五、HttpClient封装的Get请求，根据名称查队列、根据名称查交换机

  1. public class HttpUtil {
  2.  
  3.   /**
  4.    * HttpClient封装的Get请求
  5.    * 
  6.    * @param url
  7.    * @return
  8.    */
  9.   public static String get(String url) {
  10.     // 创建HttpClientBuilder
  11.     HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
  12.  
  13.     // HttpClient
  14.     CloseableHttpClient httpClient = httpClientBuilder.build();
  15.  
  16.     HttpGet httpGet = new HttpGet(url);
  17.  
  18.     RabbitConfig config = new RabbitConfig();
  19.  
  20.     final String authInfo = config.getUsername() + ":" + config.getPassword();
  21.  
  22.     System.out.println("authInfo : " + authInfo);
  23.  
  24.     try {
  25.       String encoding = DatatypeConverter.printBase64Binary("guest:guest".getBytes("utf-8"));
  26.       httpGet.setHeader("Authorization", "Basic " + encoding);
  27.     } catch (UnsupportedEncodingException e) {
  28.       e.printStackTrace();
  29.     }
  30.  
  31.     System.out.println(httpGet.getRequestLine());
  32.     try {
  33.       // 执行get请求
  34.       HttpResponse httpResponse = httpClient.execute(httpGet);
  35.       // 获取响应消息实体
  36.       HttpEntity entity = httpResponse.getEntity();
  37.  
  38.       String content = EntityUtils.toString(entity);
  39.  
  40.       // 响应状态
  41.       System.out.println("status:" + httpResponse.getStatusLine());
  42.       System.out.println("content:" + content);
  43.  
  44.       // 判断响应实体是否为空
  45.       if (entity != null) {
  46.         System.out.println("contentEncoding:" + entity.getContentEncoding());
  47.       }
  48.       return content;
  49.     } catch (IOException e) {
  50.       e.printStackTrace();
  51.     } finally {
  52.       try { // 关闭流并释放资源
  53.         httpClient.close();
  54.       } catch (IOException e) {
  55.         e.printStackTrace();
  56.       }
  57.     }
  58.  
  59.     return null;
  60.   }
  61.  
  62.   /**
  63.    * URL 解码
  64.    *
  65.    */
  66.   public static String getURLDecoderString(String str) {
  67.     String result = "";
  68.     if (null == str) {
  69.       return "";
  70.     }
  71.     try {
  72.       result = java.net.URLDecoder.decode(str, "utf-8");
  73.     } catch (UnsupportedEncodingException e) {
  74.       e.printStackTrace();
  75.     }
  76.     return result;
  77.   }
  78.  
  79.   /**
  80.    * URL 转码
  81.    *
  82.    */
  83.   public static String getURLEncoderString(String str) {
  84.     String result = "";
  85.     if (null == str) {
  86.       return "";
  87.     }
  88.     try {
  89.       result = java.net.URLEncoder.encode(str, "utf-8");
  90.     } catch (UnsupportedEncodingException e) {
  91.       e.printStackTrace();
  92.     }
  93.     return result;
  94.   }
  95.  
  96. }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/人工智能uu/article/detail/976501
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号