5

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-cliredis 附带的客户端中，控制权不会返回给客户端，所以如果我键入unsubscribe它不会去任何地方。

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

版本：

$ ./redis-server --version
Redis 服务器 v=2.6.14 sha=00000000:0 malloc=libc bits=64

4

3 回答 3

6

通过客户，我相信他们的意思是这里的客户名单：

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

于 2013-07-12T18:26:12.223 回答
3

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

于 2014-02-20T07:06:31.443 回答
0 
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;
}

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

  • 订阅：（第一个 exe）

    ./pubsub channel1

  • 发布：（第二个exe）

    ./pubsub pub channel1 hi

    在第一个 exe 中输出

    0

    回复：留言

    收到消息 -> 嗨（在频道：频道 1）

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

    0 回复：消息

    收到消息 -> 嗨（在频道：频道 1）

    取消订阅

    0

    回复：退订

于 2013-11-20T07:58:30.587 回答