发后即忘模型：简单说就是创建了任务，放置到交换器上，让应用程序继续返回工作
如：
通知---对发生事件的描述，内容可以是日志，可以报告给管理员或者程序
批处理---针对大数据集合的工作或者转换
一.告警系统
会自动将告警信息路由到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; }}