RabbitMQ(7)-发后即忘模型
RabbitMQ(7)-发后即忘模型
你我他有个梦 发表于2年前
RabbitMQ(7)-发后即忘模型
  • 发表于 2年前
  • 阅读 170
  • 收藏 1
  • 点赞 0
  • 评论 1

【腾讯云】新注册用户域名抢购1元起>>>   

摘要: 讲解这个模型下的集中java实现,例如邮件通知和任务的批处理

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

如:

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

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

一.告警系统

会自动将告警信息路由到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;
    }
}


  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
你我他有个梦
粉丝 91
博文 110
码字总数 98858
评论 (1)
chaun
:+1:
×
你我他有个梦
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: