Redis發(fā)布訂閱機制學(xué)習(xí)|kafka相關(guān)經(jīng)驗

這篇具有很好參考價值的文章主要介紹了Redis發(fā)布訂閱機制學(xué)習(xí)|kafka相關(guān)經(jīng)驗。希望對大家有所幫助。如果存在錯誤或未考慮完全的地方，請大家不吝賜教，您也可以點擊"舉報違法"按鈕提交疑問。

Redis
鍵值數(shù)據(jù)庫 key value
NoSql

第一章：差異
#1 structured 結(jié)構(gòu)化
約束 primary unique unsigned
#2 relational 關(guān)聯(lián)的
#3 SQL查詢
例 select id, name, age from tb_user where id=1
redis get user:1
mongoDB db.users.find({_id: 1})
elasticsearch GET http://localhost:9200/users/1
#4 事務(wù) ACID 基本一致無事務(wù)
鍵值類型 Redis 文檔類型 MongoDB 列類型 HBase Graph類型 Neo4j
存儲方式磁盤內(nèi)存
擴展性垂直水平
Remote Dictionary Server 遠程詞典服務(wù)器基于內(nèi)存的鍵值型NoSQL數(shù)據(jù)庫
特征：
鍵值key-value型 value支持多種不同數(shù)據(jù)結(jié)構(gòu)，功能豐富
單線程，每個命令具備原子性低延遲速度快基于內(nèi)存，IO多路復(fù)用
-支持數(shù)據(jù)持久化
-支持主從集群，分片集群
-支持多語言客戶端

安裝Redis依賴 redius是基于C語言編寫需要安裝Redis所需要的gcc依賴:
yum install -y gcc tcl
解壓: tar -zxvf redis-6.2.6.tar.gz
進入redis目錄: cd redis-6.2.6
運行編輯命令: make && make install 編譯安裝
內(nèi)容： redis-cli 命令行客戶端 redis-server
任意目錄 redis-server 端口 6379
–默認啟動方式
–指定配置啟動流程: 配置文件備份 cp redis.conf redis.conf.bck
修改redis.conf文件配置
bind 0.0.0.0 任意ip地址可以訪問
#守護進程，修改為yes后即可后臺運行 daemonize yes
#密碼，設(shè)置后訪問redis必須輸入密碼 requirepass xxxxx
其他常見配置 #監(jiān)聽端口 port
#工作目錄 dir
#數(shù)據(jù)庫數(shù)量 databases 默認有16個庫，編號0-15
#能夠使用的最大內(nèi)存 maxmemory 512mb
#日志文件默認為空不記錄日志 logfile “redis.log”
啟動 redis-server redis.conf
—開機自啟

redis客戶端
1 命令行客戶端
使用方式: redis-cli [options][commonds]
常見options -h IP地址 -p 端口 -a 訪問密碼
操作命令commonds ping 與redis服務(wù)端做心跳測試，服務(wù)端正常會返回pong
set name xxx
get name
SELECT 0 0號庫

2 圖形化客戶端

3 編程客戶端

redis常見命令
數(shù)據(jù)結(jié)構(gòu)介紹 key-value
key一般是string類型
value多類型 string hash list set sortedset geo bitmap hyperlog
redis官網(wǎng) commands 分組
命令行 help help @

redis通用命令 keys組下 @generic
KEYS: 查看符合模板的所有key(通配符) 例 KEYS * KEYS a* 以a開頭
DEL：刪除一個指定的KEY
MSET：批量插入
EXISTS: 判斷key是否存在
EXPIRE: 給key設(shè)置有效期有效期到期key被自動刪除 EXPIRE key seconds
TTL：查看KEY的剩余有效期

String類型底層都是字節(jié)數(shù)組形式存儲，編碼方式不同。
SET GET MSET MGET INCR INCRBY 自增，自增指定步長 SETNX （前提key不存在否則不執(zhí)行）
Redis的key允許多個單詞形成層級結(jié)構(gòu)，多個單詞用’:'隔開，
例: 項目名:業(yè)務(wù)名:類型:id

hash類型散列 value是無序字典，類似java中的hashmap結(jié)構(gòu)
常見命令前綴加H field

List類型雙向鏈表結(jié)構(gòu) 支持正向檢索反向檢索

java客戶端
Jedis

引入依賴

redis.clients
jedis
3.7.0
建立連接
private Jedis jedis;
@BeforeEach
void setUp(){
//建立連接
jedis = new Jedis(“xxx”, xxx);
jedis.auth(“password”); //密碼
jedis.select(0);
}

Jedis本身線程不安全頻繁創(chuàng)建和銷毀會有性能損耗推薦使用連接池
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(); //最大連接
jedisPoolConfig.setMaxIdle(); //最大空閑連接
jedisPoolConfig.setMaxIdle(); //最小空閑連接
jedisPoolConfig.setMaxWaitMillis(); //設(shè)置最長等待時間
jedisPool = new JedisPool(jedisPoolConfig, xxx, xxx, timeout, password)

public static Jedis getJedis(){
return jedisPool.getResource();
}

Redis 發(fā)布訂閱機制
簡介：
Redis 發(fā)布訂閱（Pus/Sub）是一種消息通信模式：發(fā)送者通過 PUBLISH發(fā)布消息，訂閱者通過 SUBSCRIBE 訂閱接收消息或通過UNSUBSCRIBE 取消訂閱。主要由「發(fā)布者」、「訂閱者」、「Channel」三個部分組成。
發(fā)布者和訂閱者屬于客戶端，Channel 是 Redis 服務(wù)端，發(fā)布者將消息發(fā)布到頻道，訂閱這個頻道的訂閱者則收到消息。

1 基于頻道的發(fā)布訂閱
//在redisServer中，有一個字典類型字段pubsub_channels 用來保存訂閱信息，其中key為頻道，value為訂閱該頻道的客戶端
struct redisServer{
pid_t pid;
//將頻道映射到已訂閱客戶端的列表
dict *pubsub_channels
}

2 基于模式的發(fā)布訂閱
//在redisServer中有一個pubsub_patterns屬性，該屬性表示一個鏈表，鏈表中保存著所有和模式相關(guān)的信息
struct redisServer{
list *pubsub_patterns；
}
typedef struct pubsubPattern{
client *client; – 訂閱模式客戶端
robj *pattern; --被訂閱的模式
} pubsubPattern;

需要注意的是，發(fā)布消息與監(jiān)聽消息要運行在不同的 JVM，如果使用同一個 redissonClient 發(fā)布的話，不會監(jiān)聽到自己的消息。

缺陷：
發(fā)布者不知道訂閱者是否收到發(fā)布的消息
訂閱者不知道自己是否收到了發(fā)布者發(fā)出的所有消息
發(fā)送者不能獲知訂閱者的執(zhí)行情況
沒人知道訂閱者何時開始收到消息

實現(xiàn)
生產(chǎn)者代碼

 * 發(fā)布消息到 Topic
 * @param message 消息
 * @return 接收消息的客戶端數(shù)量
public long sendMessage(String message) {
    RTopic topic = redissonClient.getTopic(CHANNEL);
    long publish = topic.publish(message);
    log.info("生產(chǎn)者發(fā)送消息成功，msg = {}", message);
    return publish;
}

消費者代碼

public void onMessage() {
  // in other thread or JVM
  RTopic topic = redissonClient.getTopic(CHANNEL);
  topic.addListener(String.class, (channel, msg) -> {
    log.info("channel: {} 收到消息 {}.",  channel, msg);
  });
}

Spring boot整合redis

消息監(jiān)聽配置
@Configuration
public class RedisSubConfig {
    public static final String SUB_KEY = "chat";//頻道channel
     * redis消息監(jiān)聽器容器
     * 可以添加多個監(jiān)聽不同話題的redis監(jiān)聽器，只需要把消息監(jiān)聽器和相應(yīng)的消息訂閱處理器綁定，該消息監(jiān)聽器
     * 通過反射技術(shù)調(diào)用消息訂閱處理器的相關(guān)方法進行一些業(yè)務(wù)處理
     * @param connectionFactory
     * @param listenerAdapter
     * @return
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //訂閱了一個頻道
        container.addMessageListener(listenerAdapter, new PatternTopic(RedisSubConfig.SUB_KEY));
        return container;
    }
    
   * 消息監(jiān)聽器適配器，綁定消息處理器，利用反射技術(shù)調(diào)用消息處理器的業(yè)務(wù)方法
    @Bean
    MessageListenerAdapter listenerAdapter(RedisReceiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }
    /**
     * redis 讀取內(nèi)容的template
     * @param connectionFactory
     * @return
     */
    @Bean
    StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }
}

接收消息

@Service
public class RedisReceiver {
    public void receiveMessage(String message) {
        System.out.println("接收消息：" + message);
    }
}

采用定時器發(fā)布消息

@EnableScheduling //開啟定時器功能
@Component
public class MessageSender {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Scheduled(fixedRate = 5000) //間隔5s 通過StringRedisTemplate對象向redis消息隊列chat頻道發(fā)布消息
    public void sendMessage(){
        stringRedisTemplate.convertAndSend("chat", "hello "+ new Date());
    }
}

kafka相關(guān)
消息頭格式
RecordHeaders(headers = [RecordHeader(key = messageType, value = [0, 0, 0, 1]), RecordHeader(key = operationCode, value = [0, 0, 0, 1]), RecordHeader(key = messageId, value = [52, 52, 52, 53, 53, 53])], isReadOnly = false)
使用java讀取消息頭

private MsgHeader parseMsgHeaders(Headers headers) {
        MsgHeader msgHeader = new MsgHeader();
        Header xxxHeader = headers.lastHeader("xxx");
        if (xxxHeader != null) {
            msgHeader.setXXX(new String(xxxHeader.value()));
        }
        return msgHeader;
    }

使用go發(fā)送消息頭文章來源地址http://www.zghlxwxcb.cn/news/detail-606628.html

headers := []sarma.RecordHeader{
    sarama.RecordHeader{
        Key: []byte("kkk"),
        Value: []byte("vvv"),
}}
msg := &sarama.ProducerMessage{
    Topic: "topic",
    Key: sarama.StringEncoder("  "),
    Value: sarama.StringEncode("  "),
    Headers: headers,
}

到了這里，關(guān)于Redis發(fā)布訂閱機制學(xué)習(xí)|kafka相關(guān)經(jīng)驗的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！