文档章节

RabbitMQ(7)-发后即忘模型

你我他有个梦
 你我他有个梦
发布于 2015/12/20 17:39
字数 1323
阅读 173
收藏 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;
    }
}


© 著作权归作者所有

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

你我他有个梦

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

评论(1)

chaun
chaun
:+1:
【Python模块】rabbitMQ

RabbitMQ介绍: 父进程与子进程间,同一父继承可以用multiprocess的Manager模块来实现数据互访。 作用:RabbitMQ是为了实现相互独立的两个进程数据互访。 应用场景:不需要立即操作的数据。比...

等你的破船
08/13
0
0
伍哥原创之安装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
center 安装 rabbitMQ

centerOs 安装 消息队列 rabbitMQ 下载 https://www.erlang-solutions.com/resources/download.html 编辑文件 centOS7 安装 使用 yum 安装, 推荐安装方式(解决依赖) 下载 rabbitMQ-server ht...

anziguoer
08/28
0
0
RabbitMQ管理插件的安装

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

openthings
2015/05/21
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Shell编程(expect同步文件、指定host和同步文件、构建文件分发系统、批量执行命令)

expect脚本同步文件 需求:自动同步文件 实验准备: A机器:192.168.248.130 B机器:192.168.248.129 实现: 1.A机器编写4.expect脚本文件,内容如下所示: #!/usr/bin/expectset passwd "...

蛋黄_Yolks
15分钟前
0
0
ppwjs之bootstrap颜色:背景颜色

<!DOCTYPT html><html><head><meta http-equiv="content-type" content="text/html; charset=utf-8" /><title>ppwjs欢迎您</title><link rel="icon" href="/favicon.ico" ......

ppwjs
16分钟前
0
0
Ubuntu与 Fedora之对比

大家好。今天我将重点介绍两个流行的Linux发行版之间的一些特性和差异; Ubuntu 18.04和Fedora 28。它们都有自己的包管理; Ubuntu使用DEB,而Fedora使用RPM,但它们都具有相同的桌面环境(GNO...

linuxprobe16
20分钟前
0
0
线性代数入门

线性代数的概念对于理解机器学习背后的原理非常重要,尤其是在深度学习领域中。它可以帮助我们更好地理解算法内部到底是怎么运行的,借此,我们就能够更好的做出决策。所以,如果你真的希望了...

牛奋Debug
昨天
2
0
开发5分钟,调试2小时 - 该如何debug?

几年来我在答疑群、论坛、公众号、知乎回答的各种问题,没有一万也有八千。其中有三分之二以上都是在帮人看报错,帮人 debug(调试代码)。 可以说,会不会 debug,有没有 debug 的意识,懂不...

crossin
昨天
3
1

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部