1、进入rabbitmq的sbin目录,进入CMD
2、输入./rabbitmq-plugins enable rabbitmq_management启用管理服务。
3、输入./rabbitmqctl start_app启动服务。
查看是否启动成功
1、浏览器访问http://localhost:15672/
下载erlang
地址如下:
http://erlang.org/download/otp_win64_20.3.exe
找到 otp_win64_20.3.exe,以管理员方式运行此文件,安装。
erlang安装完成需要配置erlang环境变量:
这两个都加入到系统盘中去
ERLANG_HOME=D:\Program Files\erl9.3
在path中添加%ERLANG_HOME%\bin;
2)安装RabbitMQ
https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.7.3
找到 rabbitmq-server-3.7.3.exe,以管理员方式运行此文件,安装。
普通java 项目的依赖 <dependency> ? ? ? ? ? ?<groupId>com.rabbitmq</groupId> ? ? ? ? ? ?<artifactId>amqp-client</artifactId> ? ? ? ? ? ?<version>4.0.3</version> ? ? ? ?</dependency>
TOPIC 模式 也就是路由模式 ? p ?提供了准确的路径 c ?提供了大概得路径 ? 来消费数据
Confirm 确认消息 用在 p -> ex 上面的 先连同 以后在上传数据,两上去发送失败就会触发这个机制
channel.confirmSelect(); // 开启监听 可能不止一个管道 channel.addConfirmListener(new ConfirmListener() { ? ? ? ? ? ?public void handleAck(long l, boolean b) throws IOException { ? ? ? ? ? ? ? ?System.out.println(l); ? ? ? ? ? ? ? ?System.out.println(b); ? ? ? ? ? ? ? ?System.out.println("确认获取..."); ? ? ? ? ? } ? ? ? ? ? ? ?public void handleNack(long l, boolean b) throws IOException { ? ? ? ? ? ? ? ?System.out.println(l); ? ? ? ? ? ? ? ?System.out.println(b); ? ? ? ? ? ? ? ?System.out.println("未获取..."); ? ? ? ? ? } ? ? ? });
retrun 消息机制
? // 执行失败的时候添加的返回数据 成的时候什么都不会 这里是它自己的bug ? ? ? ?channel.addReturnListener(new ReturnListener() { ? ? ? ? ? ?public void handleReturn(int i, String s, String s1, String s2, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException { ? ? ? ? ? ? ? ?String str = new String(bytes, "UTF-8"); ? ? ? ? ? ? ? ?System.out.println("没有路由成功的数:" + str); ? ? ? ? ? } ? ? ? });
消费端的限流
channel.basicQos(2); // 每次消费两条数据 限流策略关键代码 // 开启acKnowledge验证数据 读出数据单未消费掉 channel.basicConsume("myqosqu", false, consumer);
消费端的ACK与重回队列 (消费端)
// 手动消费 channel.basicAck(envelope.getDeliveryTag(),false); ?// 两个参数 获取每一条消费的对象id 不配量处理 / // 手动不消费 也不返回队列(false) true 是返回队列那就死循环了,一直在处理这条数据 channel.basicNack(envelope.getDeliveryTag(),false,true);
有效时间 (p)
AMQP.BasicProperties properties = new AMQP.BasicProperties() ? ? ? ? ? ? ? .builder() ? ? ? ? ? ? ? .deliveryMode(2)// 设置消息是否持久化,1: 非持久化 2:持久化 ? ? ? ? ? ? ? .contentEncoding("UTF-8") ? ? ? ? ? ? ? .expiration("10000") ? ? ? ? ? ? ? .build(); // p 端的数据 第三个参数来设置有效时间 hannel.basicPublish("myoneEx", "myoneAs", properties, smsMsg.getBytes());
死信队列
?Map<String, Object> param = new HashMap<String, Object>(); //创建一个死信队列 ? ? ? ?param.put("x-dead-letter-exchange", "mytwoEx"); ? ? ? ? ?channel.exchangeDeclare("myoneEx", BuiltinExchangeType.DIRECT); // 如果出现问题就放到第二交换机上面去(在队列上设置属性的) ? ? ? ?channel.queueDeclare("myoneQu", true, false, false, param);
springboot 的使用
-- 相关依赖 <dependency> ? ? ? ? ? ?<groupId>org.springframework.boot</groupId> ? ? ? ? ? ?<artifactId>spring-boot-starter-amqp</artifactId> ?</dependency>
这里是springboot一些注解,防止自己忘记,在此查阅了一些资料
@Bean不仅可以作用在方法上,也可以作用在注解类型上,在运行时提供注册。 ? value:name属性的别名,在不需要其他属性时使用,也就是说value 就是默认值 name:此bean 的名称,或多个名称,主要的bean的名称加别名。如果未指定,则bean的名称是带注解方法的名称。如果指定了,方法的名称就会忽略,如果没有其他属性声明的话,bean的名称和别名可能通过value属性配置 autowire :此注解的方法表示自动装配的类型,返回一个Autowire类型的枚举,我们来看一下Autowire枚举类型的概念
当使用在属性上的时候,如下代码,@Qualifier 和 @Autowired结合使用可以通过唯一Bean的id实现自动装配,因为单独的@Autowired注解实现自动装配是按照类型优先原则的,一旦IOC容器中出现了两个类型一样的Bean,@Autowired注解就会无法辨别用那个,即而报错,但是当我们加上 @Qualifier(value = "Bean的id") 的时候就可以直接通过Bean的唯一标识(id)进行装配了。
@RabbitListener注解既可以加在类上,也可以加在方法上,其作用有所不同。
正如上面讲的,当@RabbitListener注解加在类上时,表示该类是一个RabbitMQ消息监听器容器,可以包含多个带有@RabbitHandler注解的方法,用于处理不同类型的消息。例如:
@RabbitListener(queues = MQConfig.PAYQUEUE) public class MQListener{ @RabbitHandler // 这里处理多个类型的数据 ? ? public void revicePaySuccess(String orderNo){ ? ? ? ? } }
因此,@RabbitListener注解加在类上时,表示该类是一个消息监听器容器,可以包含多个处理不同类型消息的方法;而加在方法上时,表示该方法是一个消息监听器,只用于处理特定类型的消息。