java远程连接调用Rabbitmq的实例代码

发布时间 - 2026-01-11 02:22:34    点击率:

本文介绍了java远程连接调用Rabbitmq,分享给大家,希望此文章对各位有所帮助。

打开IDEA创建一个maven工程(Java就可以了)。

 

pom.xml文件如下

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>

 <groupId>com.zhenqi</groupId>
 <artifactId>rabbitmq-study</artifactId>
 <version>1.0-SNAPSHOT</version>
 <packaging>jar</packaging>

 <name>rabbitmq-study</name>
 <url>http://maven.apache.org</url>

 <properties>
  <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 </properties>

 <dependencies>
  <dependency>
   <groupId>junit</groupId>
   <artifactId>junit</artifactId>
   <version>4.12</version>
   <scope>test</scope>
  </dependency>

  <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
  <dependency>
   <groupId>com.rabbitmq</groupId>
   <artifactId>amqp-client</artifactId>
   <version>4.1.0</version>
   <exclusions>
    <exclusion>
     <groupId>org.slf4j</groupId>
     <artifactId>slf4j-api</artifactId>
    </exclusion>
   </exclusions>
  </dependency>

  <dependency>
   <groupId>org.slf4j</groupId>
   <artifactId>slf4j-log4j12</artifactId>
   <version>1.7.21</version>
  </dependency>

  <dependency>
   <groupId>commons-lang</groupId>
   <artifactId>commons-lang</artifactId>
   <version>2.6</version>
  </dependency>

 </dependencies>
</project>

为了能远程访问rabbitmq,则需要编辑 /etc/rabbitmq/rabbitmq.conf,添加以下内容。

[
  {rabbit, [{tcp_listeners, [5672]}, {loopback_users, ["asdf"]}]}
 ]

添加administrator角色

rabbitmqctl set_user_tags openstack administrator

创建抽象队列 EndPoint.java

package com.zhenqi;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


/**
 * Created by wuming on 2017/7/16.
 */
public abstract class EndPoint {

  protected Channel channel;
  protected Connection connection;
  protected String endPointName;

  public EndPoint(String endpointName) throws Exception {

    this.endPointName = endpointName;

    //创建一个连接工厂 connection factory
    ConnectionFactory factory = new ConnectionFactory();

    //设置rabbitmq-server服务IP地址
    factory.setHost("192.168.146.128");
    factory.setUsername("openstack");
    factory.setPassword("rabbitmq");
    factory.setPort(5672);
    factory.setVirtualHost("/");

    //得到 连接
    connection = factory.newConnection();

    //创建 channel实例
    channel = connection.createChannel();

    channel.queueDeclare(endpointName, false, false, false, null);
  }

  /**
   * 关闭channel和connection。并非必须,因为隐含是自动调用的。
   * @throws IOException
   */
  public void close() throws Exception{
    this.channel.close();
    this.connection.close();
  }
}

生产者Producer.java

生产者类的任务是向队列里写一条消息

package com.zhenqi;

import org.apache.commons.lang.SerializationUtils;

import java.io.Serializable;

/**
 * Created by wuming on 2017/7/16.
 */
public class Producer extends EndPoint {

  public Producer(String endpointName) throws Exception {
    super(endpointName);
  }

  public void sendMessage(Serializable object) throws Exception {
    channel.basicPublish("",endPointName, null, SerializationUtils.serialize(object));
  }
}

消费者QueueConsumer.java

消费者可以以线程方式运行,对于不同的事件有不同的回调函数,其中最主要的是处理新消息到来的事件。

package com.zhenqi;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import org.apache.commons.lang.SerializationUtils;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by wuming on 2017/7/16.
 */
public class QueueConsumer extends EndPoint implements Runnable, Consumer {

  private Logger LOG=Logger.getLogger(QueueConsumer.class);

  public QueueConsumer(String endpointName) throws Exception {
    super(endpointName);
  }

  public void handleConsumeOk(String s) {

  }

  public void handleCancelOk(String s) {

  }

  public void handleCancel(String s) throws IOException {

  }

  public void handleShutdownSignal(String s, ShutdownSignalException e) {

  }

  public void handleRecoverOk(String s) {
    LOG.info("Consumer "+s +" registered");
  }

  public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
    Map map = (HashMap) SerializationUtils.deserialize(bytes);
    LOG.info("Message Number "+ map.get("message number") + " received.");
  }

  public void run() {
    try{
      channel.basicConsume(endPointName, true,this);
    }catch(IOException e){
      e.printStackTrace();
    }
  }
}

 测试

运行一个消费者线程,然后开始产生大量的消息,这些消息会被消费者取走

package com.zhenqi;

import java.util.HashMap;

/**
 * Created by wuming on 2017/7/16.
 */
public class TestRabbitmq {

  public static void main(String[] args){
    try{
      QueueConsumer consumer = new QueueConsumer("queue");
      Thread consumerThread = new Thread(consumer);
      consumerThread.start();

      Producer producer = new Producer("queue");

      for (int i = 0; i < 100000; i++){
        HashMap message = new HashMap();
        message.put("message number", i);
        producer.sendMessage(message);
        System.out.println("Message Number "+ i +" sent.");
      }
    }catch(Exception e){
      e.printStackTrace();
    }

  }
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。


# java远程调用Rabbitmq  # java远程调用  # Java RabbitMQ的工作队列与消息应答详解  # Java RabbitMQ的持久化和发布确认详解  # Java面试高频问题之RabbitMQ系列全面解析  # Java搭建RabbitMq消息中间件过程详解  # JAVA获取rabbitmq消息总数过程详解  # Java编程rabbitMQ实现消息的收发  # java中RabbitMQ高级应用  # 创建一个  # 的是  # 给大家  # 最主要  # 回调  # 大家多多  # 就可以  # 取走  # 则需  # 远程访问  # 新消息  # exclusions  # client  # api  # slf4j  # exclusion  # amqp  # junit  # dependency  # UTF 


相关栏目: 【 网站优化151355 】 【 网络推广146373 】 【 网络技术251813 】 【 AI营销90571


相关推荐: Microsoft Edge如何解决网页加载问题 Edge浏览器加载问题修复  rsync同步时出现rsync: failed to set times on “xxxx”: Operation not permitted  济南网站建设制作公司,室内设计网站一般都有哪些功能?  javascript中的try catch异常捕获机制用法分析  googleplay官方入口在哪里_Google Play官方商店快速入口指南  Laravel如何使用Contracts(契约)进行编程_Laravel契约接口与依赖反转  如何用搬瓦工VPS快速搭建个人网站?  香港服务器租用每月最低只需15元?  Laravel怎么判断请求类型_Laravel Request isMethod用法  高防服务器如何保障网站安全无虞?  怎么用AI帮你设计一套个性化的手机App图标?  做企业网站制作流程,企业网站制作基本流程有哪些?  Laravel如何使用集合(Collections)进行数据处理_Laravel Collection常用方法与技巧  香港服务器选型指南:免备案配置与高效建站方案解析  轻松掌握MySQL函数中的last_insert_id()  Linux网络带宽限制_tc配置实践解析【教程】  如何用y主机助手快速搭建网站?  JS中对数组元素进行增删改移的方法总结  Laravel如何实现数据导出到CSV文件_Laravel原生流式输出大数据量CSV【方案】  佛山企业网站制作公司有哪些,沟通100网上服务官网?  Laravel如何集成Inertia.js与Vue/React?(安装配置)  长沙企业网站制作哪家好,长沙水业集团官方网站?  如何快速搭建安全的FTP站点?  免费网站制作appp,免费制作app哪个平台好?  大型企业网站制作流程,做网站需要注册公司吗?  Python文件操作最佳实践_稳定性说明【指导】  Laravel如何使用Gate和Policy进行授权?(权限控制)  如何彻底删除建站之星生成的Banner?  5种Android数据存储方式汇总  如何用IIS7快速搭建并优化网站站点?  如何在阿里云虚拟主机上快速搭建个人网站?  PHP正则匹配日期和时间(时间戳转换)的实例代码  北京企业网站设计制作公司,北京铁路集团官方网站?  宙斯浏览器文件分类查看教程 快速筛选视频文档与图片方法  Laravel如何实现数据导出到PDF_Laravel使用snappy生成网页快照PDF【方案】  如何快速生成可下载的建站源码工具?  Laravel如何自定义分页视图?(Pagination示例)  家族网站制作贴纸教程视频,用豆子做粘帖画怎么制作?  如何快速生成高效建站系统源代码?  米侠浏览器网页背景异常怎么办 米侠显示修复  如何基于云服务器快速搭建网站及云盘系统?  利用JavaScript实现拖拽改变元素大小  简单实现Android验证码  b2c电商网站制作流程,b2c水平综合的电商平台?  实现点击下箭头变上箭头来回切换的两种方法【推荐】  智能起名网站制作软件有哪些,制作logo的软件?  高端建站如何打造兼具美学与转化的品牌官网?  如何快速上传建站程序避免常见错误?  实例解析Array和String方法  php增删改查怎么学_零基础入门php数据库操作必知基础【教程】