package com.delqueue;  import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;  /** * 消息体定义 实现Delayed接口就是实现两个方法即compareTo 和 getDelay最重要的就是getDelay方法，这个方法用来判断是否到期…… */
public class Message implements Delayed {  private int id;  private String body; // 消息内容  private long excuteTime;// 延迟时长，这个是必须的属性因为要按照这个判断延时时长。  public int getId() {  return id;  }  public String getBody() {  return body;  }  public long getExcuteTime() {  return excuteTime;  }  public Message(int id, String body, long delayTime) {  this.id = id;  this.body = body;  this.excuteTime = TimeUnit.NANOSECONDS.convert(delayTime, TimeUnit.MILLISECONDS) + System.nanoTime();  }  // 自定义实现比较方法返回 1 0 -1三个参数  @Override  public int compareTo(Delayed delayed) {  Message msg = (Message) delayed;  return Integer.valueOf(this.id) > Integer.valueOf(msg.id) ? 1  : (Integer.valueOf(this.id) < Integer.valueOf(msg.id) ? -1 : 0);  }  // 延迟任务是否到时就是按照这个方法判断如果返回的是负数则说明到期否则还没到期  @Override  public long getDelay(TimeUnit unit) {  return unit.convert(this.excuteTime - System.nanoTime(), TimeUnit.NANOSECONDS);  }
}  

package com.delqueue;  import java.util.concurrent.DelayQueue;  public class Consumer implements Runnable {  // 延时队列 ,消费者从其中获取消息进行消费  private DelayQueue<Message> queue;  public Consumer(DelayQueue<Message> queue) {  this.queue = queue;  }  @Override  public void run() {  while (true) {  try {  Message take = queue.take();  System.out.println("消费消息id：" + take.getId() + " 消息体：" + take.getBody());  } catch (InterruptedException e) {  e.printStackTrace();  }  }  }
}  

package com.delqueue;  import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;  public class DelayQueueTest {  public static void main(String[] args) {    // 创建延时队列    DelayQueue<Message> queue = new DelayQueue<Message>();    // 添加延时消息,m1 延时3s    Message m1 = new Message(1, "world", 3000);    // 添加延时消息,m2 延时10s    Message m2 = new Message(2, "hello", 10000);    //将延时消息放到延时队列中  queue.offer(m2);    queue.offer(m1);    // 启动消费线程 消费添加到延时队列中的消息，前提是任务到了延期时间   ExecutorService exec = Executors.newFixedThreadPool(1);  exec.execute(new Consumer(queue));  exec.shutdown();  }
}  

package com.test.delayqueue;
/** * 具体执行相关业务的业务类 * @author whd * @date 2017年9月25日 上午12:49:32 */
public class DelayOrderWorker  implements Runnable {  @Override  public void run() {  // TODO Auto-generated method stub  //相关业务逻辑处理  System.out.println(Thread.currentThread().getName()+" do something ……");  }
}  

package com.test.delayqueue;  import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;  /** * 延时队列中的消息体将任务封装为消息体 *  * @author whd * @date 2017年9月25日 上午12:48:30 * @param <T> */
public class DelayOrderTask<T extends Runnable> implements Delayed {  private final long time;  private final T task; // 任务类，也就是之前定义的任务类  /** * @param timeout *            超时时间(秒) * @param task *            任务 */  public DelayOrderTask(long timeout, T task) {  this.time = System.nanoTime() + timeout;  this.task = task;  }  @Override  public int compareTo(Delayed o) {  // TODO Auto-generated method stub  DelayOrderTask other = (DelayOrderTask) o;  long diff = time - other.time;  if (diff > 0) {  return 1;  } else if (diff < 0) {  return -1;  } else {  return 0;  }  }  @Override  public long getDelay(TimeUnit unit) {  // TODO Auto-generated method stub  return unit.convert(this.time - System.nanoTime(), TimeUnit.NANOSECONDS);  }  @Override  public int hashCode() {  return task.hashCode();  }  public T getTask() {  return task;  }
}  

package com.test.delayqueue;  import java.util.Map;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;  /** * 延时队列管理类，用来添加任务、执行任务 *  * @author whd * @date 2017年9月25日 上午12:44:59 */
public class DelayOrderQueueManager {  private final static int DEFAULT_THREAD_NUM = 5;  private static int thread_num = DEFAULT_THREAD_NUM;  // 固定大小线程池  private ExecutorService executor;  // 守护线程  private Thread daemonThread;  // 延时队列  private DelayQueue<DelayOrderTask<?>> delayQueue;  private static final AtomicLong atomic = new AtomicLong(0);  private final long n = 1;  private static DelayOrderQueueManager instance = new DelayOrderQueueManager();  private DelayOrderQueueManager() {  executor = Executors.newFixedThreadPool(thread_num);  delayQueue = new DelayQueue<>();  init();  }  public static DelayOrderQueueManager getInstance() {  return instance;  }  /** * 初始化 */  public void init() {  daemonThread = new Thread(() -> {  execute();  });  daemonThread.setName("DelayQueueMonitor");  daemonThread.start();  }  private void execute() {  while (true) {  Map<Thread, StackTraceElement[]> map = Thread.getAllStackTraces();  System.out.println("当前存活线程数量:" + map.size());  int taskNum = delayQueue.size();  System.out.println("当前延时任务数量:" + taskNum);  try {  // 从延时队列中获取任务  DelayOrderTask<?> delayOrderTask = delayQueue.take();  if (delayOrderTask != null) {  Runnable task = delayOrderTask.getTask();  if (null == task) {  continue;  }  // 提交到线程池执行task  executor.execute(task);  }  } catch (Exception e) {  e.printStackTrace();  }  }  }  /** * 添加任务 *  * @param task * @param time *            延时时间 * @param unit *            时间单位 */  public void put(Runnable task, long time, TimeUnit unit) {  // 获取延时时间  long timeout = TimeUnit.NANOSECONDS.convert(time, unit);  // 将任务封装成实现Delayed接口的消息体  DelayOrderTask<?> delayOrder = new DelayOrderTask<>(timeout, task);  // 将消息体放到延时队列中  delayQueue.put(delayOrder);  }  /** * 删除任务 *  * @param task * @return */  public boolean removeTask(DelayOrderTask task) {  return delayQueue.remove(task);  }
}  

package com.delqueue;  import java.util.concurrent.TimeUnit;  import com.test.delayqueue.DelayOrderQueueManager;
import com.test.delayqueue.DelayOrderWorker;  public class Test {  public static void main(String[] args) {  DelayOrderWorker work1 = new DelayOrderWorker();// 任务1  DelayOrderWorker work2 = new DelayOrderWorker();// 任务2  DelayOrderWorker work3 = new DelayOrderWorker();// 任务3  // 延迟队列管理类，将任务转化消息体并将消息体放入延迟对列中等待执行  DelayOrderQueueManager manager = DelayOrderQueueManager.getInstance();  manager.put(work1, 3000, TimeUnit.MILLISECONDS);  manager.put(work2, 6000, TimeUnit.MILLISECONDS);  manager.put(work3, 9000, TimeUnit.MILLISECONDS);  }  }  

OK 这就是项目中的具体使用情况，当然具体内容被忽略，整体框架就是这样，还有这里使用java的延时队列但是这种方式是有问题的如果如果down机则会出现任务丢失，所以也可以考虑使用mq、redis来实现

2、mq实现延迟消息

在rabbitmq 3.5.7及以上的版本提供了一个插件（rabbitmq-delayed-message-exchange）来实现延迟队列功能。同时插件依赖Erlang/OPT 18.0及以上。
插件源码地址：
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange

插件下载地址：
https://bintray.com/rabbitmq/community-plugins/rabbitmq_delayed_message_exchange

安装：

进入插件安装目录
{rabbitmq-server}/plugins/（可以查看一下当前已存在的插件）
下载插件

rabbitmq_delayed_message_exchangewget https://bintray.com/rabbitmq/community-plugins/download_file?file_path=rabbitmq_delayed_message_exchange-0.0.1.ez

（如果下载的文件名称不规则就手动重命名一下如：

rabbitmq_delayed_message_exchange-0.0.1.ez）

启用插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

关闭插件

rabbitmq-plugins disable rabbitmq_delayed_message_exchange

插件使用

通过声明一个x-delayed-message类型的exchange来使用delayed-messaging特性
x-delayed-message是插件提供的类型，并不是rabbitmq本身的，发送消息的时候通过在header添加”x-delay”参数来控制消息的延时时间

直接在maven工程的pom.xml文件中加入

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

接下来在 application.properties 文件中加入redis配置：

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

也很简单，代码如下：

package com.mq.rabbitmq;import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitMqConfig {private String host;private int port;private String userName;private String password;@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(host,port);cachingConnectionFactory.setUsername(userName);cachingConnectionFactory.setPassword(password);cachingConnectionFactory.setVirtualHost("/");cachingConnectionFactory.setPublisherConfirms(true);return cachingConnectionFactory;}@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());return rabbitTemplate;}public String getHost() {return host;}public void setHost(String host) {this.host = host;}public int getPort() {return port;}public void setPort(int port) {this.port = port;}public String getUserName() {return userName;}public void setUserName(String userName) {this.userName = userName;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;}
}

package com.mq.rabbitmq;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class QueueConfig {@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange("test_exchange", "x-delayed-message",true, false,args);}@Beanpublic Queue queue() {Queue queue = new Queue("test_queue_1", true);return queue;}@Beanpublic Binding binding() {return BindingBuilder.bind(queue()).to(delayExchange()).with("test_queue_1").noargs();}
}

实现消息发送

package com.mq.rabbitmq;import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.text.SimpleDateFormat;
import java.util.Date;@Service
public class MessageServiceImpl {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMsg(String queueName,String msg) {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("消息发送时间:"+sdf.format(new Date()));rabbitTemplate.convertAndSend("test_exchange", queueName, msg, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setHeader("x-delay",3000);return message;}});}
}

x-delay

在这里我设置的延迟时间是3秒。

消息消费者

package com.mq.rabbitmq;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.text.SimpleDateFormat;
import java.util.Date;
@Component
public class MessageReceiver {@RabbitListener(queues = "test_queue_1")public void receive(String msg) {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("消息接收时间:"+sdf.format(new Date()));System.out.println("接收到的消息:"+msg);}
}

直接在main方法里运行Spring Boot程序,Spring Boot会自动解析 MessageReceiver 类的。

接下来只需要用Junit运行一下发送消息的接口即可。

package com.mq.rabbitmq;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {@Autowiredprivate MessageServiceImpl messageService;@Testpublic void send() {messageService.sendMsg("test_queue_1","hello i am delay msg");}
}

运行完后，可以看到如下信息：

消息发送时间:2018-05-03 12:44:53
3秒钟后，Spring Boot控制台会输出:
消息接收时间:2018-05-03 12:44:56
接收到的消息:hello i am delay msg