03 redis發布訂閱和mq(redis也可以實現隊列，為什么還要用rabitmq或者kafka)

監聽器 

redis:listener-買粉絲ntainer：定義消息監聽，method：監聽消息執行的方法，serializer：序列化，topic：監聽主題（可以理解為隊列名稱）

3.3代碼實現

1.定義短信消息對象SmsMessageVo

public class SmsMessageVo implements Serializable {

//id

private Integer smsId;    //手機號

private String mobile;    //類型，1：驗證碼 2：訂單通知

private Byte type;    //短信創建時間

private Date createDate;    //短信消息處理時間

private Date processTime;    //短信狀態，1：未發送 2：發送成功 3：發送失敗

private Byte status;    //短信內容

private String 買粉絲ntent;    //省略setter和getter方法

...123456789101112131415161718192021222324

2.定義消息隊列發送對象SendMessage

//SendMessage.javapublic class SendMessage {    private RedisTemplate<String, Object> redisTemplate;    public RedisTemplate<String, Object> getRedisTemplate() {        return redisTemplate;

}    public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {        this.redisTemplate = redisTemplate;

}    public void sendMessage(String channel, Serializable message) {

redisTemplate.買粉絲nvertAndSend(channel, message);

}

}12345678910111213141516171819202122

3.發送消息

String smsContent = templateToContent(template.getContent(),

regMsgCode);

SmsMessageVo smsMessageVo = new SmsMessageVo();

smsMessageVo.setMobile(mobile);

smsMessageVo.setType((byte) SmsType.VERIFICATION.getType());

smsMessageVo.setChannelId(1);

smsMessageVo.setContent(smsContent);

smsMessageVo.setCreateDate(new Date());

smsMessageVo.setStatus((byte) SmsSendStatus.TO_SEND.getType());

smsMessageVo.setTemplateId(1);

//先把待發送的短信存入數據庫

SmsQueue smsQueue = new SmsQueue();

BeanUtils.買粉絲pyProperties(smsQueue, smsMessageVo);

smsQueueService.addSmsQueue(smsQueue);

//異步發送短信到redis隊列

sendMessage.sendMessage(Constants.REDIS_QUEUE_SMS_WEB, smsMessageVo);

//Constants.REDIS_QUEUE_SMS_WEB = "sms_queue_web_online",和applicationContext-redis中topic配置一樣123456789101112131415161718192021

4.監聽消息

//SmsMessageDelegateListener.java@Component("smsMessageDelegateListener")public class SmsMessageDelegateListener {

@Autowired

private SmsQueueService smsQueueService;    //監聽Redis消息

public void handleMessage(Serializable message){        if(message instanceof SmsMessageVo){

SmsMessageVo messageVo = (SmsMessageVo) message;            //發送短信

SmsSender smsSender = SmsSenderFactory.buildEMaySender();

smsSender.setMobile(messageVo.getMobile());

smsSender.setContent(messageVo.getContent());            boolean sendSucc = false;            //判斷短信類型

//驗證碼短信

if(messageVo.getType() == (byte)SmsType.VERIFICATION.getType()){

sendSucc = smsSender.send();

}            if(!sendSucc){                return;

}            // 異步更新短信表狀態為發送成功

final Integer smsId = messageVo.getSmsId();

Executor executor = Executors.newSingleThreadExecutor();

executor.execute(new Runnable() {                public void run() {

SmsQueue smsQueue = new SmsQueue();

smsQueue.setSmsId(smsId);

smsQueue.setStatus((byte)SmsSendStatus.SEND.getType());

smsQueue.setProcessTime(new Date());

smsQueueService.updateSmsQueue(smsQueue);

}

});

}

}

}123456789101112131415161718192021222324252627282930313233343536373839404142434445

4.總結

下面使用一張流程圖，來總結Redis消息隊列和短信服務。 

閱讀全文

Kafka，Mq和Redis作為消息隊列使用

kafka是個日志處理緩沖組件，在大數據信息處理中使用。和傳統的消息隊列相比較簡化了隊列結構和功能，以流形式處理存儲（持久化）消息（主要是日志）。日志數據量巨大，處理組件一般會處理不過來，所以作為緩沖層的kafka，支持巨大吞吐量。為了防止信息丟失，其消息被調用后不直接丟棄，要多存儲一段時間，等過期時間過了才丟棄。這是mq和redis不能具備的。主要特點如下:巨型存儲量: 支持TB甚至PB級別數據。高吞吐，高IO：一般配置的服務器能實現單機每秒100K以上消息的傳輸。消息分區，分布式消費：能保消息順序傳輸。 支持離線數據處理和實時數據處理。Scale out：支持在線水平擴展，以支持更大數據處理量

redis只是提供一個高性能的、原子操作內存鍵值對，具有高速訪問能力，可用做消息隊列的存儲，但是不具備消息隊列的任何功能和邏輯，要作為消息隊列來實現的話，功能和邏輯要通過上層應用自己實現。

我們以RabbitMQ為例介紹。它是用Erlang語言開發的開源的消息隊列，支持多種協議，包括AMQP，XMPP, SMTP, STOMP。適合于企業級的開發。

MQ支持Broker構架，消息發送給客戶端時需要在中心隊列排隊。對路由，負載均衡或者數據持久化都有很好的支持。

還有ActiveMq，ZeroMq等。功能基本上大同小異。并發吞吐TPS比較，ZeroMq 最好，RabbitMq 次之， ActiveMq 最差。

原文：

很赞哦!（26）