Redis取消订阅

Question

Redis支持PUBSUB。订阅很容易：

redis 127.0.0.1:6379> subscribe foo
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "foo"
3) (integer) 1

然而，似乎无法取消订阅，因为在订阅时，服务器不接受命令。例如，在附带redis的redis-cli客户端中，控制权不会返回到客户端，因此如果我键入unsubscribe，它将无处可去。

这似乎是文档、函数或PEBKAC问题中的一个明显错误。出了什么问题？

版本： $ ./redis-server --version Redis server v=2.6.14 sha=00000000:0 malloc=libc bits=64

- djechlin

---

你正在使用哪个版本的Redis？ - Homer6

@Homer6 版本 2.6.14 - djechlin

Answer 1

我认为“client”指的是这里列出的客户端列表：http://redis.io/clients。作为一个使用 hiredis 客户端的人，我认为这篇文章中的建议：

一旦客户端进入订阅状态，它就不应该发出任何其他命令，除了额外的 SUBSCRIBE、PSUBSCRIBE、UNSUBSCRIBE 和 PUNSUBSCRIBE 命令。

适用于那些客户端。 http://redis.io/commands/subscribe 上的这个页面。Redis-cli 就是其中之一。所以，这条评论不是针对 redis-cli 用户的指令。

相反，redis-cli 阻塞等待总线上的消息（只有通过 ctrl+c 取消订阅）。如果您使用不同的客户端（或更具体地说，如果您正在实现一个客户端），我的猜测是，您必须遵守这个约定，使它处于订阅状态（尽管客户端不一定会阻塞）。

我认为文档可能需要更清晰地澄清这一点；然而，文档位于服务器本身而不是 redis-cli 应用程序上。但是，您可以在文档存储库中进行调整并提交拉取请求：https://github.com/antirez/redis-doc/blob/master/commands/subscribe.md。

- Homer6

Answer 2

实际上，PSUBSCRIBE 与 SUBSCRIBE 一样会阻塞所有后续命令，因此你不能向服务器发送任何订单，只能将目光投向等待感兴趣的频道传入消息。这种荒谬的行为让我头晕目眩。 然而，如果您尝试通过telnet（例如telnet localhost 6379）而不是redis-cli提示与redis交互，那么一切都会没问题。享受它吧。

- user1611552

Answer 3

An example of publishing subscribing and unsubscribing in c++.




#include <signal.h>
#include <iostream>
#include <stdio.h>
#include <stdlib.h>
#include <string>
#include <hiredis/hiredis.h>
#include <hiredis/async.h>
#include <hiredis/adapters/libevent.h>
#include <boost/thread/thread.hpp>

using namespace std;

struct event_base* base;
std::string CHANNEL("");


void subCallback(redisAsyncContext *c, void *r, void *privdata) ;

void unSubscribe(redisAsyncContext* _redisContext){

  std::string input;

  while(1)
  {
    cin>>input;
    if(input.compare("unsub") == 0)
      break;
    sleep(5);
  }

  std::string command("unsubscribe ");
  command.append(CHANNEL);

  cout<<
    redisAsyncCommand(_redisContext, 
        subCallback, 
        (char*)"unsub", command.c_str())<<endl; 
}

void subCallback(redisAsyncContext *c, void *r, void *privdata) {

  redisReply *reply = (redisReply*)r;
  if (reply == NULL){
    cout<<"Response not recev"<<endl; 
    return;
  }
  if(reply->type == REDIS_REPLY_ARRAY & reply->elements == 3)
  {
    if(strcmp( reply->element[0]->str,"subscribe") != 0)
    {
      cout<<"Reply for:  "<<reply->element[0]->str<<endl;
      if(strcmp( reply->element[0]->str,"unsubscribe") == 0)
      {
        exit(1);
      }

      cout<<"Message received -> "<<
        reply->element[2]->str<<"( on channel : "<<reply->element[1]->str<<")"<<endl;
    }
  }
}


void pubCallback(redisAsyncContext *c, void *r, void *privdata) {

  redisReply *reply = (redisReply*)r;
  if (reply == NULL){
    cout<<"Response not recev"<<endl; 
    return;
  }
  cout<<"message published"<<endl;
  redisAsyncDisconnect(c);
}

void connectCallback(const redisAsyncContext *c, int status) {
  if (status != REDIS_OK) {
    cout<<"Error in connect: %s\n"<< c->errstr<<endl;
    return;
  }
  cout<<"Connected to redis server..."<<endl;
}

void disconnectCallback(const redisAsyncContext *c, int status) {
  if (status != REDIS_OK) {
    cout<<"Error in disconnect: %s\n"<< c->errstr<<endl;
    return;
  }
  cout<<"Disconnected...\n"<<endl;
}

int main(int argv, char** args)
{
  string processName(args[1]);

  signal(SIGPIPE, SIG_IGN);
//  struct event_base*
    base = event_base_new();

  redisAsyncContext* 
    _redisContext = redisAsyncConnect("10.0.0.30", 6379);

  if (_redisContext->err) {
    /* Let context leak for now... */
    cout<<"Error: "<< _redisContext->errstr<<endl;
    return 1;
  }

  redisLibeventAttach(_redisContext,base);
  //redisAsyncSetConnectCallback(_redisContext,connectCallback);
  //redisAsyncSetDisconnectCallback(_redisContext,disconnectCallback);

  if(processName.compare("pub") == 0)
  {
    string command ("publish ");
    command.append(args[2]);
    command.append (" ");
    command.append(args[3]);

    cout<<
      redisAsyncCommand(_redisContext, 
          pubCallback, 
          (char*)"pub", command.c_str())<<endl; 

  }
  else if(processName.compare("sub") == 0)
  {
    boost::thread unsubscribe(&unSubscribe, _redisContext);

    string command ("subscribe ");
    command.append(args[2]);

    CHANNEL.append(args[2]);

    cout<<
      redisAsyncCommand(_redisContext, 
          subCallback, 
          (char*)"sub", command.c_str())<<endl; 

  }
  else
    cout<<"Try pub or sub"<<endl;

  event_base_dispatch(base);

  return 0;
}

可以运行代码来测试订阅、发布和取消订阅：

• 订阅：（第一步执行）

  ./pubsub channel1
  

• 发布：（第二步执行）

  ./pubsub pub channel1 hi
  

  在第一步执行中的输出：

  0

  回复消息：message

  接收到的消息 -> hi（在频道channel1上）

• 现在取消订阅第一步执行，输入unsub在第一步执行控制台中。最终输出：

  0 回复消息：message

  接收到的消息 -> hi（在频道channel1上）

  unsub

  0

  回复消息：取消订阅

- Nik