文档章节

RabbitMQ(7)-发后即忘模型

你我他有个梦
 你我他有个梦
发布于 2015/12/20 17:39
字数 1323
阅读 171
收藏 1
点赞 0
评论 1

发后即忘模型:简单说就是创建了任务,放置到交换器上,让应用程序继续返回工作

如:

通知---对发生事件的描述,内容可以是日志,可以报告给管理员或者程序

批处理---针对大数据集合的工作或者转换

一.告警系统

会自动将告警信息路由到critical队列或者rate_limit队列上

1.告警系统生产者:

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

public class AlertWarningProducer {
    private static final String EXCHANGE_NAME = "alerts";
    private static final String ROUTING_KEY = "critical.alert";
  //private static final String ROUTING_KEY = "alert.rate_limt";
    public static void main(String[] argv) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("liuzhaoqiang128");
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setPort(5672);

            connection = factory.newConnection();
            channel = connection.createChannel();

            channel.exchangeDeclare(EXCHANGE_NAME, "topic",true);

            String message = "critical content!!!";
            //String message = "rate_limt content!!!";
            //消息发布
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes());
            System.out.println(" [x] Sent '" + ROUTING_KEY + "':'" + message + "'");

        }
        catch  (Exception e) {
            e.printStackTrace();
        }
        finally {
            if (connection != null) {
                try {
                    connection.close();
                }
                catch (Exception ignore) {}
            }
        }
    }
}

2.告警系统消费者

import com.rabbitmq.client.*;

import javax.mail.*;
import javax.mail.Message.RecipientType;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.TimeoutException;

public class AlertWarningConsumer {
    private  final static  String EMAIL_RECIPIENTS="*******@163.com";//接收者
    private  final static  String EMAIL_SENDER="*******@163.com";//发送者
    private  final static  String EXCHANGE="alerts";
    private  final static  String TYPE="topic";
    private  final static  String QUEUE1="critical";
    private  final static  String QUEUE2="rate_limt";
    private  final static  String ROUTING_KEY1="critical.*";
    private  final static  String ROUTING_KEY2="*.rate_limt";
    /**
     *
     * @param recipients    接收人
     * @param subject       发送主题
     * @param msg       发送消息内容
     * @throws IOException
     * @throws MessagingException
     */
    public static void sendEmail(String recipients, String subject, Object msg) throws IOException, MessagingException {
        final Properties props = new Properties();
        /*
         * 可用的属性: mail.store.protocol / mail.transport.protocol / mail.host /
         * mail.user / mail.from
         */
        // 表示SMTP发送邮件,需要进行身份验证
        props.put("mail.smtp.auth", "true");
        props.put("mail.smtp.host", "smtp.163.com");
        // 发件人的账号
        props.put("mail.user", EMAIL_SENDER);
        // 访问SMTP服务时需要提供的密码
        props.put("mail.password", "qiangzai123");

        // 构建授权信息,用于进行SMTP进行身份验证
        Authenticator authenticator = new Authenticator() {
            @Override
            protected PasswordAuthentication getPasswordAuthentication() {
                // 用户名、密码
                String userName = props.getProperty("mail.user");
                String password = props.getProperty("mail.password");
                return new PasswordAuthentication(userName, password);
            }
        };
        // 使用环境属性和授权信息,创建邮件会话
        Session mailSession = Session.getInstance(props, authenticator);
        // 创建邮件消息
        MimeMessage message = new MimeMessage(mailSession);
        // 设置发件人
        InternetAddress form = new InternetAddress(
                props.getProperty("mail.user"));
        message.setFrom(form);

        // 设置收件人
        InternetAddress to = new InternetAddress(recipients);
        message.setRecipient(RecipientType.TO, to);

        // 设置邮件标题
        message.setSubject(subject);
        // 设置邮件的内容体{"message":"告警消息邮件发送"}
        message.setContent(msg, "application/json;charset=UTF-8");
        // 发送邮件
        Transport.send(message);
    }
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        Connection connection = null;

        try {

            factory.setPort(5672);
            factory.setHost("liuzhaoqiang128");
            factory.setUsername("admin");
            factory.setPassword("admin");
            connection = factory.newConnection();
            //创建连接
            final Channel channel = connection.createChannel();
            //声明交换器队列绑定等信息
            channel.exchangeDeclare(EXCHANGE, TYPE, true);
            channel.queueDeclare(QUEUE1, false, false, false, null);
            channel.queueBind(QUEUE1,EXCHANGE,ROUTING_KEY1);

            channel.queueDeclare(QUEUE2, false, false, false, null);
            channel.queueBind(QUEUE2,EXCHANGE,ROUTING_KEY2);

            Consumer rate_limit_notify = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String subject = "rate_limit Alert";
                    String msg = new String(body,"UTF-8");
                    try {
                        sendEmail(EMAIL_RECIPIENTS, subject, msg);
                    } catch (MessagingException e) {
                        e.printStackTrace();
                    }
                    System.out.println("send alert E-mail!Alert text:Recipients: " + EMAIL_RECIPIENTS+" subject: "+subject);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };

            Consumer critical_notify = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String subject = "Critical Alert";
                    String msg = new String(body,"UTF-8");
                    try {
                        sendEmail(EMAIL_RECIPIENTS, subject, msg);
                    } catch (MessagingException e) {
                        e.printStackTrace();
                    }
                    System.out.println("send alert E-mail!Alert text:Recipients: " + EMAIL_RECIPIENTS+" subject: "+subject);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            //消息消费
            channel.basicConsume(QUEUE1,false,"critical",critical_notify);
            channel.basicConsume(QUEUE2,false,"rate_limit",rate_limit_notify);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

二.并行处理

如上图所示,可以看到一个名为"upload-pictures"的交换器,下方绑定了几个queue

其中一个使用场景,有三个任务,图片尺寸调整、奖励用户积分以及通知所有用户,这几个任务需要并行处理,因为这几个任务毫不相干,不需要等待上一个任务完成,现在想增加一个新类型的任务,譬如日志记录,只需定义新的队列绑定到upload-pictures交换器上即可

例如图片上传通过RbbitMQ发布一条元数据信息,然后让他真正执行任务的异步工作者去处理即可

部分代码实现如下:

1.信息发布者

import com.rabbitmq.client.*;


public class UploadPicturePublish {
    private static final String EXCHANGE_NAME = "upload_pictures";
    public static void main(String[] argv) {
        ConnectionFactory factory = new ConnectionFactory();
        Connection connection = null;
        Channel channel = null;
        try {
            factory.setHost("liuzhaoqiang129");
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setPort(5672);

            connection = factory.newConnection();
            channel = connection.createChannel();

            channel.exchangeDeclare(EXCHANGE_NAME,"fanout",true,false,false,null);
            String jsonMessage = "{\"image_id\":123456,\"user_id\":123456,\"image_path\":\"pic_jpg\"}";
            channel.basicPublish(EXCHANGE_NAME,"", MessageProperties.PERSISTENT_TEXT_PLAIN,jsonMessage.getBytes("UTF-8"));
            System.out.println("Send message:" + jsonMessage);
            Thread.sleep(10000);
        }
        catch  (Exception e) {
            e.printStackTrace();
        }
        finally {
            if (connection != null) {
                try {
                    connection.close();
                }
                catch (Exception ignore) {}
            }
        }
    }
}

2.消费者

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;

import java.io.IOException;

public class UploadPictureConsumer {
    private static final String EXCHANGE_NAME = "upload_pictures";
    private static final String QUEUE="add_points";
    private static final String QUEUE1="resize_pictures";
    private static final String QUEUE2="notify-friends";
    private static final String QUEUE3="logs";
    public static void main(String[] argv) {
        ConnectionFactory factory = new ConnectionFactory();
        Connection connection = null;
        Channel channel = null;
        try {
            factory.setHost("liuzhaoqiang129");
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setPort(5672);

            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true, false, false, null);
            
            //积分消费
            Consumer add_point = integrationConsumer(channel);
            channel.basicConsume(QUEUE,false,add_point);
            Thread.sleep(10000);
        }
        catch  (Exception e) {
            e.printStackTrace();
        }
        finally {
            if (connection != null) {
                try {
                    connection.close();
                }
                catch (Exception ignore) {}
            }
        }
    }

    public static Consumer integrationConsumer(final Channel channel){
        channel.queueDeclare(QUEUE,false,false,false,null);
        channel.queueBind(QUEUE,EXCHANGE_NAME,"");
        //这里有一个应用场景就是可以获取元数据中的用户ID,这样可以从数据库中获取所有联系人,去通知联系人
        //这里的回调函数可以动态注册
        Consumer add_point = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String jsonString = new String(body,"UTF-8");
                String message = JSON.parseObject(jsonString).getString("user_id");
                if(message.equals("quit")){
                    channel.basicCancel(consumerTag);
                }
                System.out.println("Adding to points to user:    " + message);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };

        return add_point;
    }
    public static Consumer resizePictures(final Channel channel){
        channel.queueDeclare(QUEUE1,false,false,false,null);
        channel.queueBind(QUEUE1,EXCHANGE_NAME,"");
        Consumer resize_pictures = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String jsonString = new String(body,"UTF-8");
                String image_id = JSON.parseObject(jsonString).getString("image_id");
                String image_path = JSON.parseObject(jsonString).getString("image_path");
                System.out.println("Adding to points to user:    " + image_id+"---"+image_path);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        return resize_pictures;
    }
}


© 著作权归作者所有

共有 人打赏支持
你我他有个梦

你我他有个梦

粉丝 92
博文 110
码字总数 98858
作品 0
昌平
程序员
加载中

评论(1)

chaun
chaun
:+1:
伍哥原创之安装RabbitMQ

测试环境:CentOS 6.2 首先用root身份登录测试环境 1,安装erlang (R15B01) 安装完成以后,执行erl看是否能打开eshell,用’halt().’退出,注意后面的点号,那是erlang的结束符。 如下: ...

alex.wu
2012/08/06
0
0
消息中间件—RabbitMQ(集群原理与搭建篇)

摘要:实际生产应用中都会采用消息队列的集群方案,如果选择RabbitMQ那么有必要了解下它的集群方案原理 一般来说,如果只是为了学习RabbitMQ或者验证业务工程的正确性那么在本地环境或者测试...

癫狂侠
05/25
0
0
RabbitMQ管理插件的安装

先安装rabbitmq-server这里就不写了,之前有篇文章里有相关步骤:RabbitMQ的安装与配置 如果/etc/rabbitmq不存在会报如下错误: Error: {cannotwriteenabledpluginsfile,”/etc/rabbitmq/ena...

openthings
2015/05/21
0
0
消息中间件—RabbitMQ(初探篇)

文章摘要:本篇文章为RabbitMQ的入门文章,不像其他一些程序代码和应用实战性的文章会带着大家从一个“Hello World”的简单例子出发,在该篇幅中主要给大家讲下RabbitMQ消息队列的起源、为何...

癫狂侠
05/23
0
0
RabbitMQ消息中间件介绍

一、基础介绍 随着分布式应用的发展消息队列中间件成为C/S架构中解耦的一个重要环节,传统的消息传输模型中,C端发出消息,S端必须在线,否则将无法继续进行,而在拥有消息中间件的模型下消息...

qiao645
06/21
0
0
RabbitMQ 内存控制 硬盘控制

一、内存控制: vmmemoryhigh_watermark 该值为内存阈值,默认为0.4。意思为物理内存的40%。40%的内存并不是内存的最大的限制,它是一个发布的节制,当达到40%时Erlang会做GC。最坏的情况是使...

andrewniu
05/10
0
0
Centos7部署RabbitMQ 3.7.6

简介 MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。消息传递指的是...

阿dai
07/10
0
0
安装 RabbitMQ 消息队列

1、安装运行环境 yum install erlang 2、安装RabbitMQ wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.3.5/rabbitmq-server-3.3.5-1.noarch.rpm yum install rabbitmq-server-3.......

李伟铭k
07/09
0
0
RabbitMQ安装(CentOS 7 64位)

一、安装Erlang 详细的安装介绍在这里(https://www.erlang-solutions.com/downloads/download-erlang-otp) wget http://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rp......

nibilly
2015/04/28
0
0
RabbitMQ 的安装与工作模式

RabbitMQ 概念: 交换机(exchange type)把消息推送到队列的方法: fanout:不处理路由键,转发到所有绑定的队列上 direct:处理路由键,必须完全匹配,即路由键字符串相同才会转发 topic:路由...

求学ing
2014/11/11
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

sap netweaver developer studio安装svn插件

问题 我现在在Sap的IDE(netweaver developer studio)上面安装svn插件。 步骤 确定IDE使用的eclipse版本 Help→About SAP NetWeaver Developer Studio→Installation Details→Features→F...

亚林瓜子
5分钟前
0
0
Spring Cloud云服务架构 - commonservice-config配置服务搭建

1. 介绍 Spring Cloud Config为分布式系统中的外部配置提供服务器和客户端支持。使用Config Server,您可以在所有环境中管理应用程序的外部属性。客户端和服务器上的概念映射与Spring Enviro...

itcloud
6分钟前
1
0
大数据开发学习的内容介绍,成都大数据培训机构哪里好?

大数据开发培训已经成为了越来越多人的选择,大数据开发工程师也是各公司争相争夺的金领人才之一了,在当今科技发展非常迅速的社会里,越来越多人把职业规划投向了大数据开发。这里为大家整理...

加米谷大数据
10分钟前
2
0
函数

函数 函数是Python中最主要也是最重要的代码组织和复用手段。作为最重要的原则,如果你要重复使用相同或非常类似的代码,就需要写一个函数。通过给函数起一个名字,还可以提高代码的可读性。...

火力全開
11分钟前
0
0
gulp-webserverf启动服务,局域网无法访问

如题,gulp-server启动的服务,只能本机访问,局域网通过ip无法访问; 启动的其它项目,均可以访问成功; 网上资源,很多说什么防火墙之类的问题,都无果; 只需要给启动服务添加参数即可, ...

littleFaye
13分钟前
0
0
RabbitMQ实战:5种模式和示例

应用RabbitMQ的5种队列 一、简单队列 P:消息的生产者 C:消息的消费者 红色:队列 生产者实现思路: 创建连接工厂ConnectionFactory,设置服务地址127.0.0.1,端口号5672,设置用户名、密码...

spinachgit
14分钟前
0
0
mysql常见报错标号对应原因以及处理方法

mysql常见报错标号以及对应解决方法 报错标号 报错现象 解决方法 原因 1449 Cause: java.sql.SQLException: The user specified as a definer ('authplat_dev'@'%') does not exist 在控制台...

ChinaHYF
16分钟前
0
0
Java 监控系统技术选型

(1)操作系统监控 Sigar oshi (2)Tomcat监控 JMX 日志 (3)Oracle监控 日志 直连SQL查询 基于Druid连接池 (4)拓扑图 jtopo http://www.jtopo.com/demo/statictis.html...

cccyb
18分钟前
1
0
解决IDEA中moduel配置了maven依赖可是依然不能使用依赖中的类

POM.xml中明明配置了依赖,也开启了maven的 auto-import 下面的刷新maven也没用: 直到使用下面的解决办法才使依赖生效: IDEA打开右侧 maven projects 点击顶部的M图表(看下图) 出现如下对...

颖辉小居
19分钟前
0
0
Nginx proxy pass路由转发简单用法

一,在nginx中配置proxy_pass时的加不加/的问题要注意proxy_pass后的url最后的/当加上了/,相当于是绝对根路径,则nginx不会把location中匹配的路径部分代理走如果没有/,则会把匹配的...

binhu
19分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部