来源： 程序员开发正遭 AI 「革命」_CSDN资讯的博客-CSDN博客

【CSDN 编者按】AI，一个可以让人充满无限想象力的技术。近年来，随着 GPT-3、BERT 等大语言模型的崛起，以及昨日最新 Stable Diffusion 2.0 开源文本图像模型的发布，也让很多「所想即所得」的应用场景逐渐成为现实。反观其背后，技术人在推动 AI 落地的过程中，包括自身的开发习惯、方式等也正被 AI 重置。

作者 | 卞安 责编 | 屠敏

出品 | CSDN（ID：CSDNnews）

AI 作画，作为一个局部领域的产品方向，在基于机器学习算法基础上，为世界打开了一扇大门，随着 AI 作画类软件的大放异彩，基于 AI 进行内容生产话题也越来越受到关注。

作为一名程序员，我们又应该怎么来看待 AI 的这些成果对于开发方式的影响，在此，笔者有幸受邀为各位小伙伴分享一下个人的看法。

在美国科罗拉多州艺术博览会上基于 Midjourney 获奖的 AI 画作《太空歌剧院》

AI 将改写程序员的开发方式

AI 的底层算法一般是基于神经网络进行建模，通过在神经元上的数学算法不断的训练数据调整建模，使误差向最小靠近的过程。这是一种非常酷的思维方式，它使得计算机可以通过算法寻找到数据变化的趋势。在此基础上发展起来的生成对抗网络，使得计算机学习和模仿数字化内容变的更加可信，这些数字化内容不但包括文本（新闻和小说生成）、图像（AI绘图与视频），3D 模型（AI 生成游戏场景），也包括编程代码和文档。

虽然我们目前还没有能够走到直接通过 AI 来生成完整的项目的地步，但在 AI 大发展的背景下，如何高效、快速、更好的达到编程目标，正在成为解决问题中一个重要的考虑因素。

我们可以对比 AI 的工作原理，反思编程工作中的痛点，预测一些潜在的可能性，也许未来这些可能性会极大的推动我们的编程和工作方式。

从工作形态的进化趋势上来看，预期会分为三个阶段：

第一阶段：编码工作的辅助智能化
这一阶段的主要体现是：通过搜集、归类、梳理编程开发项目中大量的可复用部分（如算法、功能函数或模块）形成数据集，通过对输入意向和数据集进行训练和学习的方式，形成优良的辅助功能，提升程序员的开发效率。

在具体实践上，一些企业已经推出了相关工具类产品：

GitHub Copilot : 通过语音输入指令让 IDE 自动提示并生成代码。

Copilot 是 GitHub 今年早些时候推出的人工智能编程辅助工具，2022 年 6 月 22 日已正式上线（https://github.com/features/copilot），定价每月 10 美元（约 66.9 元人民币）或每年 100 美元（约 669 元人民币），对学生用户和流行开源项目的维护者免费提供。

经过数十亿行代码的训练，它可以将自然语言提示转化为数十种语言的编码建议，支持 Python、JavaScript、TypeScript、Java、Ruby 和 Go 等编程语言。根据今年 GitHub Universe 开发者大会报告给出的数据，Copilot 已经通过基于 AI 的编码建议，帮助全球开发者的工作效率提高了 55%。

2. GodeGeeX：通过输入文字描述让 IDE 自动提示并生成代码。

与此相近的一款国产工具叫做 GodeGeeX（https://models.aminer.cn/codegeex/），来自清华大学知识工程实验证（KEG），CodeGeeX 是一款具有 130 亿参数的多语言代码生成模型，采用华为 MindSpore 框架实现，在鹏城实验室“鹏城云脑II”中的 192 个节点（共 1536 个国产昇腾 910 AI 处理器）上训练而成。

截至 2022 年 6 月 22 日，CodeGeeX 历时两个月在 20 多种编程语言的代码语料库（>8500 亿 Token）上预训练得到。它可以跟据你的描述，或者上下文自动生成一段代码，和以往的代码补全功能是完成不同的。目前支持 Python、C++、Java、JavaScript、Go 五种主流编程语言，而且在准确度上表现较好。

除此之外，还有 Kite、Codota、DeepCode 等一些基于机器学习的 AI 代码生成工具，本质上都是通过大量的代码训练，来智能预测出需要生成的代码，加快编程的效率。

总体来说，这一阶段的产品推动者不需要面向业务，只需要考虑面向程序员，可以达到辅助程序员提升业务开发速度的目标，也可以较好的通过工具化产品形成规范，解决企业的代码风格化统一问题。未来可能会广泛的成为程序员工作中的标配，程序员将不会再花大量的精力在高复用度的具体代码函数设计，而更多的把精力放在业务模块的设计实现。

第二阶段：业务模块的智能生成
这一阶段的主要体现是：通过搜集、归类、梳理相关领域业务开发项目中大量的通用业务（如游戏研发中的登录模块、热更新模块、聊天模块、战斗模块等业务逻辑）形成数据集，通过对业务设计意向和数据集进行训练和学习的方式，自动的进行业务模块前后端框架生成，优化工作流程，提升业务模块的产品设计迭代和研发速度。

我们可以通过一个实例工具软件来理解这个过程，Adobe 推出过的网页三剑客之一的 Dreamweave，其中提供了大量的模块生成功能，比如通过设定数据集和表单的控件对应关系，就可以生成前后端的代码，实际提供一个可以运行的业务模块。

当然，这是由没有结合人工智能辅助的情况，如果基于新的智能化生成，那我们在开发时，智能化生成模块会自动跟据上下文来提示是否需要生成相应的业务前后端模块，我们只需要再跟据生成的模块进行微调或扩展即可达到我们的要求。

下图是笔者使用 DreamWeaver 来生成一个简单的登录模块，只需要设定一下数据表字段，即可生成表单，完成登录判断和跳转功能。

又或者在游戏开发领域。作者曾在企业负责技术中台设计时，推动过面向各游戏项目组通用业务模块的抽象与开发，比如热更新、原生接口调用，甚至包括游戏战斗机制等大量的通用业务，对于每一个项目几乎都是需要的，只是因为技术栈或者界面的不同而需要重复的开发，这些工作往往耗时耗力。

站在一个更高的维度，这些业务完全可以基于机器学习训练结合需求变化被快速地进行调整而满足各个项目的使用。所以，基于企业的技术中台，把这些通用业务的统一框架和智能化定制与人工智能结合，基于主流开发软件、引擎工具或平台工具推出可实际操作的软件或插件。从而形成更好的开发工作流是非常有意义的。

这个阶段目前是各领域企业争相降本增效的方向之一，程序员的工作正在进一步从通用业务开发中解放，而把更多的把精力放在具体产品的需求上。

第三阶段：产品原型的智能生成
这一阶段主要体现是：通过对产品研发工作流中可智能化生成的部分进行整合和优化，达到能够根据用户需求自动的生成具备完整功能逻辑和效果表现的产品项目原型的目标，帮助产品经理快速看到期望的结果。

这里以几个案例加以说明：

一、市场上大量自助建网站，电商小程序类的平台。

这些平台直接面向用户方，提供给用户可以自助订制化的业务模块并生成最终产品的服务。如腾讯云的自助式建站：

这些平台虽然具备直接生成最终产品的能力，但其底层逻辑主要是基于一定数量的前后端模板和风格模板进行组合，通过固化的需求条目选项得到产品各模板模块的组合结果，产品的领域有限、形态有限、规模有限，目前大部分不具备通过对用户需求分析和产品形态数据集进行机器学习来生成目标产品的能力。

在数字人研发企业里，目前已经在通过对人脸照片，动作视频采用人物脸部识别，动作识别技术来建立模型数据集，通过机器学习对三维模型重建和自动动作绑定，快速化生产高精度数字人，这种方式与以往通过美术建模，手动绑定骨骼及人物动作的工作流有较大差别，它可以大大的加快数字人的生成效率、规模和精度。

在 GTC 大会上，黄仁勋宣布推出一个全方位的虚拟化身平台 —— Omniverse Avatar（来源：https://developer.nvidia.com/nvidia-omniverse-platform/ace）

二、在一些游戏方向，如剧情文字类游戏（AVG）领域，开发者也在通过剧本生成，二次元立绘生成类软件，通过脚本进行完整的游戏产品生产尝试，可以将原本需要几个月才能完成的剧情文字游戏，在短短几个小时内生成出来。

当然，这些方面的业务相对简单，但随着人工智能的发展，相信不久的未来，一个软件产品设计师可以通过提供一些简单的需求描述得到一个完整可运行的产品DEMO。它所要做的，是通过不断的修改需求描述来不断提交，得到期望的产品原型。

总结

最后，我想说的是，内容生成的算法目前在一些方面有了较大突破，但在软件开发领域，特别是开发者工具领域，我们也需要相关企业和创业者不断的结合这些突破，落地到具体的工具软件产品中来提升开发者的工作流程和体验，推出更好的产品形态，谁能够在未来提供更好的智能化内容生成体验，也就会在未来更受欢迎。

作者介绍：卞安，游戏引擎技术专家，CSDN 博客专家，独立软件创业者。从事游戏研发行业近二十年，孜孜不倦地学习和思考引擎研发技术和工具链设计，曾任端游企业引擎技术总监，Cocos 游戏引擎产品总监。目前在学习游戏引擎与 AI 的结合，致力于开发基于 Python 语言的全栈式开发工具软件产品《PyMe》。

————————————————
版权声明：本文为CSDN博主「CSDN资讯」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/csdnnews/article/details/128125034

来源： redis之mq实现发布订阅模式 – 云天 – 博客园

• 示例代码-github

概述

Redis不仅可作为缓存服务器，还可用作消息队列，本示例演示如何使用redis实现发布/订阅消息队列。

• 在Redis中，发布者没有将消息发送给特定订阅者的程序。相反，发布的消息被描述为通道，而不知道(如果有的话)可能有哪些订阅者。
• 订阅者表示对一个或多个主题感兴趣，只接收感兴趣的消息，而不知道(如果有的话)发布者是什么。
• 发布者和订阅者的这种解耦可以实现更大的可伸缩性和更动态的网络拓扑。

代码实现

redis实现mq的存储方式很多，可以使用list,zset及stream，这些数据的存储结构决定了怎么消费问题（消息是一次使用、允许多次使用、允许多端消息等），比如使用list，我们可以使用leftPush插入消息，使用rightPop消费消息，实现一条消息一次消息，可以参考与以示例代码：

    @Test
    public void testMq() {
        for (int i = 0; i < 10; i++) {
            redisTemplate.opsForList().leftPush("task-queue", "data" + i);
            log.info("插入了一个新的任务==>{}", "data" + i);
        }
        String taskId = redisTemplate.opsForList().rightPop("task-queue").toString();
        log.info("处理成功，清除任务==>{}", taskId);
    }

1.配置代码RedisConfig.java

package demo.data.mqRedis.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport {

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * redisTemplate 序列化使用的jdkSerializeable, 存储二进制字节码, 所以自定义序列化类,方便调试redis
     *
     * @param redisConnectionFactory
     * @return
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {

        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();

        //使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());

        //使用StringRedisSerializer来序列化和反序列化redis的ke
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());

        //开启事务
        redisTemplate.setEnableTransactionSupport(true);

        redisTemplate.setConnectionFactory(redisConnectionFactory);

        return redisTemplate;
    }

    @Bean
    MessageListenerAdapter messageListener() {
        return new MessageListenerAdapter(new RedisMessageSubscriber());
    }

    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter listenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, topic());

        return container;
    }

    @Bean
    MessagePublisher redisPublisher() {
        return new RedisMessagePublisher(redisTemplate, topic());
    }

    @Bean
    ChannelTopic topic() {
        return new ChannelTopic("messageQueue");
    }
}

2.定义消息发布接口MessagePublisher.java

package demo.data.mqRedis.config;

public interface MessagePublisher {
    void publish(String message);
}

3.发布方实现RedisMessagePublisher.java

package demo.data.mqRedis.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;

/**
 * 消息发布方
 */
public class RedisMessagePublisher implements MessagePublisher {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private ChannelTopic topic;

    public RedisMessagePublisher(
            RedisTemplate<String, Object> redisTemplate, ChannelTopic topic) {
        this.redisTemplate = redisTemplate;
        this.topic = topic;
    }

    public void publish(String message) {
        redisTemplate.convertAndSend(topic.getTopic(), message);
    }
}

4.消息接收方RedisMessageSubscriber.java

package demo.data.mqRedis.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * 消息订阅方
 */
@Service
@Slf4j
public class RedisMessageSubscriber implements MessageListener {

    public static List<String> messageList = new ArrayList<>();

    public void onMessage(Message message, byte[] pattern) {
        messageList.add(message.toString());
        log.info("订阅方接收到了消息==>{}", message.toString());
    }
}

5.最后贴上application.yml配置

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password:

查看运行结果

1.编写测试用例试发布消息TestRedisMQ.java

package demo.data.mqRedis;

import demo.data.mqRedis.config.RedisMessagePublisher;
import lombok.extern.slf4j.Slf4j;
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;

import java.util.UUID;

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class TestRedisMQ {

    @Autowired
    RedisMessagePublisher redisMessagePublisher;

    @Test
    public void testMq() {
        String message = "Message " + UUID.randomUUID();
        redisMessagePublisher.publish(message);
    }
}

2.运行结果

2019-09-05 15:51:33.931  INFO 10772 --- [    container-2] d.d.m.config.RedisMessageSubscriber      : 订阅方接收到了消息==>"Message c95959bf-6c30-4801-bc80-0e1e3c9f81bc"

订阅方成功接收到消息了

资料

• 示例代码-github
• redis实现mq的方案以及stream的应用
• 发布-订阅模式

关于消息队列的使用

来源： 关于消息队列的使用 – Ruthless – 博客园

一、消息队列概述
消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。目前使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ

二、消息队列应用场景
以下介绍消息队列在实际应用中常用的使用场景。异步处理，应用解耦，流量削锋和消息通讯四个场景。

2.1异步处理
场景说明：用户注册后，需要发注册邮件和注册短信。传统的做法有两种 1.串行的方式；2.并行方式
a、串行方式：将注册信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端。

b、并行方式：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间

假设三个业务节点每个使用50毫秒钟，不考虑网络等其他开销，则串行方式的时间是150毫秒，并行的时间可能是100毫秒。
因为CPU在单位时间内处理的请求数是一定的，假设CPU1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次（1000/150）。并行方式处理的请求量是10次（1000/100）
小结：如以上案例描述，传统的方式系统的性能（并发量，吞吐量，响应时间）会有瓶颈。如何解决这个问题呢？

引入消息队列，将不是必须的业务逻辑，异步处理。改造后的架构如下：

按照以上约定，用户的响应时间相当于是注册信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了两倍。

2.2应用解耦
场景说明：用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用库存系统的接口。如下图：

传统模式的缺点：假如库存系统无法访问，则订单减库存将失败，从而导致订单失败，订单系统与库存系统耦合

如何解决以上问题呢？引入应用消息队列后的方案，如下图：

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功
库存系统：订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作
假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦

2.3流量削锋
流量削锋也是消息队列中的常用场景，一般在秒杀或团抢活动中使用广泛。
应用场景：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列。
a、可以控制活动的人数
b、可以缓解短时间内高流量压垮应用

用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面。
秒杀业务根据消息队列中的请求信息，再做后续处理

2.4日志处理
日志处理是指将消息队列用在日志处理中，比如Kafka的应用，解决大量日志传输的问题。架构简化如下

日志采集客户端，负责日志数据采集，定时写受写入Kafka队列
Kafka消息队列，负责日志数据的接收，存储和转发
日志处理应用：订阅并消费kafka队列中的日志数据

2.5消息通讯
消息通讯是指，消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等
点对点通讯：

客户端A和客户端B使用同一队列，进行消息通讯。

聊天室通讯：

客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果。

以上实际是消息队列的两种消息模式，点对点或发布订阅模式。模型为示意图，供参考。

三、消息中间件示例
3.1电商系统

消息队列采用高可用，可持久化的消息中间件。比如Active MQ，Rabbit MQ，Rocket Mq。
（1）应用将主干逻辑处理完成后，写入消息队列。消息发送是否成功可以开启消息的确认模式。（消息队列返回消息接收成功状态后，应用再返回，这样保障消息的完整性）
（2）扩展流程（发短信，配送处理）订阅队列消息。采用推或拉的方式获取消息并处理。
（3）消息将应用解耦的同时，带来了数据一致性问题，可以采用最终一致性方式解决。比如主数据写入数据库，扩展应用根据消息队列，并结合数据库方式实现基于消息队列的后续处理。

3.2日志收集系统

分为Zookeeper注册中心，日志收集客户端，Kafka集群和Storm集群（OtherApp）四部分组成。
Zookeeper注册中心，提出负载均衡和地址查找服务
日志收集客户端，用于采集应用系统的日志，并将数据推送到kafka队列
Kafka集群：接收，路由，存储，转发等消息处理
Storm集群：与OtherApp处于同一级别，采用拉的方式消费队列中的数据

四、JMS消息服务
讲消息队列就不得不提JMS 。JMS（JAVA Message Service，java消息服务）API是一个消息服务的标准/规范，允许应用程序组件基于JavaEE平台创建、发送、接收和读取消息。它使分布式通信耦合度更低，消息服务更加可靠以及异步性。
在EJB架构中，有消息bean可以无缝的与JM消息服务集成。在J2EE架构模式中，有消息服务者模式，用于实现消息与应用直接的解耦。

4.1消息模型
在JMS标准中，有两种消息模型P2P（Point to Point）,Publish/Subscribe(Pub/Sub)。

4.1.1 P2P模式

P2P模式包含三个角色：消息队列（Queue），发送者(Sender)，接收者(Receiver)。每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。

P2P的特点
每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)
发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
接收者在成功接收消息之后需向队列应答成功
如果希望发送的每个消息都会被成功处理的话，那么需要P2P模式。

4.1.2 Pub/Sub模式

包含三个角色主题（Topic），发布者（Publisher），订阅者（Subscriber） 多个发布者将消息发送到Topic，系统将这些消息传递给多个订阅者。

Pub/Sub的特点
每个消息可以有多个消费者
发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息
为了消费消息，订阅者必须保持运行的状态
为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。
如果希望发送的消息可以不被做任何处理、或者只被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用Pub/Sub模型。

4.2消息消费
在JMS中，消息的产生和消费都是异步的。对于消费来说，JMS的消息者可以通过两种方式来消费消息。
（1）同步
订阅者或接收者通过receive方法来接收消息，receive方法在接收到消息之前（或超时之前）将一直阻塞；

（2）异步
订阅者或接收者可以注册为一个消息监听器。当消息到达之后，系统自动调用监听器的onMessage方法。

JNDI：Java命名和目录接口,是一种标准的Java命名系统接口。可以在网络上查找和访问服务。通过指定一个资源名称，该名称对应于数据库或命名服务中的一个记录，同时返回资源连接建立所必须的信息。
JNDI在JMS中起到查找和访问发送目标或消息来源的作用。

五、常用消息队列

一般商用的容器，比如WebLogic，JBoss，都支持JMS标准，开发上很方便。但免费的比如Tomcat，Jetty等则需要使用第三方的消息中间件。本部分内容介绍常用的消息中间件（Active MQ,Rabbit MQ，Zero MQ,Kafka）以及他们的特点。

5.1 ActiveMQ
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

ActiveMQ特性如下：
⒈ 多种语言和协议编写客户端。语言: Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议： OpenWire,Stomp REST,WS Notification,XMPP,AMQP
⒉ 完全支持JMS1.1和J2EE 1.4规范 （持久化，XA消息，事务)
⒊ 对Spring的支持，ActiveMQ可以很容易内嵌到使用Spring的系统里面去，而且也支持Spring2.0的特性
⒋ 通过了常见J2EE服务器（如 Geronimo,JBoss 4,GlassFish,WebLogic)的测试，其中通过JCA 1.5 resource adaptors的配置，可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上
⒌ 支持多种传送协议：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
⒍ 支持通过JDBC和journal提供高速的消息持久化
⒎ 从设计上保证了高性能的集群，客户端-服务器，点对点
⒏ 支持Ajax
⒐ 支持与Axis的整合
⒑ 可以很容易得调用内嵌JMS provider，进行测试

5.2 Kafka
Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群机来提供实时的消费。

Kafka是一种高吞吐量的分布式发布订阅消息系统，有如下特性：
通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。（文件追加的方式写入数据，过期的数据定期删除）
高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒数百万的消息
支持通过Kafka服务器和消费机集群来分区消息
支持Hadoop并行数据加载
Kafka相关概念
Broker
Kafka集群包含一个或多个服务器，这种服务器被称为broker[5]
Topic
每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
Partition
Parition是物理上的概念，每个Topic包含一个或多个Partition.
Producer
负责发布消息到Kafka broker
Consumer
消息消费者，向Kafka broker读取消息的客户端。
Consumer Group
每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。
一般应用在大数据日志处理或对实时性（少量延迟），可靠性（少量丢数据）要求稍低的场景使用。

来源： .Net Core里面appsettings.json的使用_65号腕的博客-CSDN博客

从.NET Core开始，项目的配置文件就换成了appsettings.json，以下整理了她的常见用法。

1. IConfiguration
IConfiguration是用来读取配置的接口，她在Microsoft.Extensions.Configuration.Abstractions包里面，命名空间为Microsoft.Extensions.Configuration。通常需要在使用的地方将她作为依赖注入，对于ASP.NET Core来说，已经默认实现了基于Json的配置，即这次要用到的appsettings.json。

2. 读取连接字符串
appsettings.json这样定义：

{
“ConnectionStrings”: {
“DbConn”: “Server=.;Database=Db;Trusted_Connection=True;”
}
}
1
2
3
4
5
调用IConfiguration的GetConnectionString方法即可

public class Startup
{
public Startup(IConfiguration configuration)
{
string connString = configuration.GetConnectionString(“DbConn”);

}
}
1
2
3
4
5
6
7
8
3. 读取单值
对于如下配置

{
“App”: {
“Name”: “Bigname65”,
“Version”: “v1.0”
}
}
1
2
3
4
5
6
如果想读取Name的值，可以有以下四种方式，各级之间用冒号连接：

string appName = configuration[“App:Name”];
appName = configuration.GetValue<string>(“App:Name”);
appName = configuration.GetSection(“App:Name”).Value;
appName = configuration.GetSection(“App”)[“Name”];
1
2
3
4
4. 读取对象
还是上面的配置文件，如果有以下C#类，应该如何直接从配置文件，读取一个实例呢？

public class AppInfo
{
public string Name { get; set; }
public string Version { get; set; }
}
1
2
3
4
5
IConfiguration有一个Bind的方法，她在Microsoft.Extensions.Configuration.Binder包里面，命名空间也是Microsoft.Extensions.Configuration，ASP.NET Core项目已经默认包含了，有以下两种写法：

AppInfo app = new AppInfo();
configuration.GetSection(“App”).Bind(app);

app = new AppInfo();
configuration.Bind(“App”, app);
1
2
3
4
5
5. IOptions
对于以上的AppInfo类以及配置，有可能需要在项目的多个地方使用到，虽然使用IConfiguration也可以，但是会产生很多重复代码。这时我们可以使用IOptions接口，她在Microsoft.Extensions.Options包里面，而且ASP.NET Core项目也已经默认包含了.

首先在Startup的ConfigureServices里面添加如下代码，这样就直接把AppInfo和配置文件关联起来了：

public void ConfigureServices(IServiceCollection services)
{
services.Configure<AppInfo>(this.Configuration.GetSection(“App”));
}
1
2
3
4
需要的时候直接使用IOptions<AppInfo>即可：

public class HomeController : Controller
{
AppInfo _app;
public HomeController(IOptions<AppInfo> appOpt)
{
_app = appOpt.Value;
}

public IActionResult Index()
{
return Content($”{_app.Name}:{_app.Version}”);
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
6. 多配置文件
ASP.NET Core项目默认会包含两个文件appsettings.json和appsettings.Development.json。看名字，后者肯定是用在开发环境的。

对于同一个配置字段，如果同时存在于appsettings.json和appsettings.Development.json两个文件当中，系统会优先取后者里面的。

正因为如此，我们还可以添加更多的配置文件,将特定环境的值（如连接字符串）放在特定的配置文件里面，比如appsettings.QA.json用于QA环境，appsettings.UAT.json用于UAT环境等。

那么不同的环境如何跟配置文件关联起来的呢？

对于开发环境，Visual Studio的项目属性中可以看到环境变量的设置。

部署以后则是在Web.config里面。

通过该环境变量则可指定配套的json文件。

6.1 发布时指定配置文件
如下图，在Visual Studio中可以看到有多个配置文件，但如果我们希望发布之后只包含需要的文件应该怎么做呢？

首先，双击项目,或者用记事本打开项目文件夹下面的.csproj文件, 添加以下代码，目的是根据不同的环境变量移除其他的配置文件。

<Project Sdk=”Microsoft.NET.Sdk.Web”>
….
<ItemGroup Condition=” ‘$(EnvironmentName)’ == ‘Development'”>
<Content Remove=”appsettings.QA.json” />
<Content Remove=”appsettings.UAT.json” />
</ItemGroup>
<ItemGroup Condition=” ‘$(EnvironmentName)’ == ‘UAT'”>
<Content Remove=”appsettings.QA.json” />
<Content Remove=”appsettings.Development.json” />
</ItemGroup>
<ItemGroup Condition=” ‘$(EnvironmentName)’ == ‘QA'”>
<Content Remove=”appsettings.UAT.json” />
<Content Remove=”appsettings.Development.json” />
</ItemGroup>
</Project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
接着就是发布，发布之前会生成Profile文件，如下图，我们生成了两个Profile，分别对应QA环境和UAT环境：

打开其中的QA.pubxml文件，添加如下代码：

<Project>
<PropertyGroup>
….
<EnvironmentName>QA</EnvironmentName>
</PropertyGroup>
</Project>
1
2
3
4
5
6
同理UAT.pubxml则添加环境变量为UAT。

最后发布，在发布目录的web.config会自动包含对应的环境变量，同时剔除了不需要的配置文件。

7. console/winform
那么如何在控制台程序或者Winform项目中使用appsettings.json配置文件呢？

首先需要添加包：Microsoft.Extensions.Configuration.Json

然后使用以下代码即可得到IConfiguration了

IConfiguration config = new ConfigurationBuilder()
.SetBasePath(Directory.GetCurrentDirectory())
.AddJsonFile(“appsettings.json”)
.Build();
1
2
3
4
最后附上源代码：

https://gitee.com/bigname65/csharp-practise/tree/master/UseAppsettings
————————————————
版权声明：本文为CSDN博主「65号腕」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/weixin_46295080/article/details/115219255

来源： 干掉RedisHelper，请这样用分布式缓存 – 掘金

namespace Microsoft.Extensions.Caching.Distributed;

/// <summary>
/// Represents a distributed cache of serialized values.
/// </summary>
public interface IDistributedCache
{
    /// <summary>
    /// Gets a value with the given key.
    /// </summary>
    byte[]? Get(string key);

    /// <summary>
    /// Gets a value with the given key.
    /// </summary>
    Task<byte[]?> GetAsync(string key, CancellationToken token = default(CancellationToken));

    void Set(string key, byte[] value, DistributedCacheEntryOptions options);

    /// <summary>
    /// Sets the value with the given key.
    /// </summary>
    Task SetAsync(string key, byte[] value, DistributedCacheEntryOptions options, CancellationToken token = default(CancellationToken));

    /// <summary>
    /// Refreshes a value in the cache based on its key, resetting its sliding expiration timeout (if any).
    /// </summary>
    void Refresh(string key);

    /// <summary>
    /// Refreshes a value in the cache based on its key, resetting its sliding expiration timeout (if any).
    /// </summary>
    Task RefreshAsync(string key, CancellationToken token = default(CancellationToken));

    /// <summary>
    /// Removes the value with the given key.
    /// </summary>
    void Remove(string key);

    /// <summary>
    /// Removes the value with the given key.
    /// </summary>
    Task RemoveAsync(string key, CancellationToken token = default(CancellationToken));
}
复制代码

 builder.Services.AddStackExchangeRedisCache(options =>
     {
         options.Configuration = builder.Configuration.GetConnectionString("MyRedisConStr");
         options.InstanceName = "SampleInstance";
     });
复制代码

// 1.启用Options以使用IOptions
services.AddOptions();
// 2.注入配置自定义配置，可以通过IOptions<T>注入到需要使用该配置的地方
services.Configure(setupAction);
// 3.注入一个单例IDistributedCache的实现类RedisCache
services.Add(ServiceDescriptor.Singleton<IDistributedCache, RedisCache>());
复制代码

private readonly SemaphoreSlim _connectionLock = new SemaphoreSlim(initialCount: 1, maxCount: 1);

[MemberNotNull(nameof(_cache), nameof(_connection))]
private void Connect()
{
    CheckDisposed();
    if (_cache != null)
    {
        Debug.Assert(_connection != null);
        return;
    }

    _connectionLock.Wait();
    try
    {
        if (_cache == null)
        {
            if (_options.ConnectionMultiplexerFactory == null)
            {
                if (_options.ConfigurationOptions is not null)
                {
                    _connection = ConnectionMultiplexer.Connect(_options.ConfigurationOptions);
                }
                else
                {
                    _connection = ConnectionMultiplexer.Connect(_options.Configuration);
                }
            }
            else
            {
                _connection = _options.ConnectionMultiplexerFactory().GetAwaiter().GetResult();
            }

            PrepareConnection();
            _cache = _connection.GetDatabase();
        }
    }
    finally
    {
        _connectionLock.Release();
    }

    Debug.Assert(_connection != null);
}
复制代码

public interface IDistributedCachePlus : IDistributedCache
{
    bool LockRelease(string key, byte[] value);

    bool LockTake(string key, byte[] value, TimeSpan expiry);
}
复制代码

services.Add(ServiceDescriptor.Singleton<IDistributedCachePlus, RedisCachePlus>());
复制代码

来源： Redis的五种数据类型的简单介绍和使用 – DingYu – 博客园

1.准备工作：

1.1在Linux下安装Redis 

https://www.cnblogs.com/dddyyy/p/9763098.html

　　1.2启动Redis

先把root/redis的redis.conf放到 /usr/local/redis/bin/目录下

使用vi 修改redis.conf 把daemonize no 变成daemonize yes

启动 ./redis-server redis.conf

测试一下 ps -ef|grep redis

　　　1.3连接客户端

2.第一种类型：String Key-Value

set key value 设置一个key 值为 value 

get key 获得key值得value 

注意：redis中的Key和Value时区分大小写的，命令不区分大小写， redis是单线程 不适合存储大容量的数据

incr key      —对应的value 自增1,如果没有这个key值 自动给你创建创建 并赋值为1

decr key     —对应的value 自减1

注意：自增的value是可以转成数字的

3.第二种类型：Hash：key-filed-value

相当于1个key 对应一个map

hset key filed value 设置值

hget key filed 　获取值

4.第三种类型：List

List 有顺序可重复

lpush list 1  2  3  4 从左添加元素

rpush list 1 2 3 4    从右添加元素

lrange list 0 -1 (从0 到-1 元素查看：也就表示查看所有)

lpop list （从左边取，删除）

rpop list  (从右边取，删除)

5.第四种类型 ：Set

Set 无顺序，不能重复

sadd set1 a b c d d (向set1中添加元素) 元素不重复

smembers set1 （查询元素）

srem set1 a （删除元素）

6.第五种类型：SortedSet（zset）

有顺序，不能重复

适合做排行榜 排序需要一个分数属性

zadd zset1 9 a 8 c 10 d 1 e   （添加元素 zadd key score member ）

(ZRANGE key start stop [WITHSCORES])(查看所有元素：zrange key  0  -1  withscores)

如果要查看分数，加上withscores.

zrange zset1 0 -1 (从小到大)

zrevrange zset1 0 -1 (从大到小)

zincrby zset2 score member (对元素member 增加 score)

127.0.0.1:6379> zadd zset1 8 a 4 b 5 c 1 d
(integer) 4
127.0.0.1:6379> zrange zset1 0 -1 
1) "d"
2) "b"
3) "c"
4) "a"
127.0.0.1:6379> zadd zset1 9 a
(integer) 0
127.0.0.1:6379> zrange zset1 0 -1 
1) "d"
2) "b"
3) "c"
4) "a"
127.0.0.1:6379> zrange zset1 0 -1 withscores
1) "d"
2) "1"
3) "b"
4) "4"
5) "c"
6) "5"
7) "a"
8) "9"
127.0.0.1:6379> zrevrange zset1 0 -1
1) "a"
2) "c"
3) "b"
4) "d"
127.0.0.1:6379> zincrby zset1 1 a
"10"
127.0.0.1:6379> zrevrange zset1 0 -1 withscores
1) "a"
2) "10"
3) "c"
4) "5"
5) "b"
6) "4"
7) "d"
8) "1"

7.Key命令

expire key second  (设置key的过期时间)

ttl key (查看剩余时间)（-2 表示不存在，-1 表示已被持久化，正数表示剩余的时间）

persist key (清除过期时间，也即是持久化 持久化成功体提示 1 不成功0)。

del key: 删除key

select 0 表示：选择0号数据库。默认是0号数据库

来源： c#使用csredis操作redis – 小y – 博客园

现在流行的redis连接客户端有StackExchange.Redis和ServiceStack.Redis，为什么选择csredis而不是这两个？

• .net 最有名望的 ServiceStack.Redis 早已沦为商业用途，在 .NETCore 中使用只能充值；
• 后来居上的 StackExchange.Redis 虽然能用，但线上各种 Timeout 错误把人坑到没脾气，两年多两年多两年多都不解决，最近发布的 2.0 版本不知道是否彻底解决了底层。
• csredis支持.net40/.net45/.netstandard2.0，基本上满足了常见运行平台，而上面两个基本已经放弃.net40了。
• csredis所有方法名与redis-cli保持一持，很容易上手！！！

环境：

• redis6.0.6
• window 10
• vs2019
• csredis3.6.5（https://github.com/2881099/csredis）
• .net4/.net 4.5/.netcore 3.1

csredis 源码地址： https://github.com/2881099/csredis

以windows服务安装Redis方法：

下载Redis服务安装包：https://github.com/tporadowski/redis/releases。

下载完成后直接点击.exe下一步下一步OK。安装完后我们会在windows服务中找到Redis Service服务。注意启动服务后在进行相关测试。

1.在.net项目中引入CSRedisCore

包安装命令：

Install-Package CSRedisCore -Version 3.6.5

2.使用：

//初始化RedisHelper对象
var csredis = new CSRedis.CSRedisClient("127.0.0.1:6379, defaultDatabase = 0, poolsize = 500, ssl = false, writeBuffer = 10240");
RedisHelper.Initialization(csredis);


//-------------字符串(string)----------------
// 添加字符串键-值对
csredis.Set("hello", "1");
csredis.Set("world", "2");
csredis.Set("hello", "3");

// 根据键获取对应的值
csredis.Get("hello");

// 移除元素
csredis.Del("world");

/*    数值操作    */
csredis.Set("num-key", "24");

// value += 5
csredis.IncrBy("num-key",5); 
// output -> 29

// value -= 10
csredis.IncrBy("num-key", -10); 
// output -> 19

/*    字节串操作    */
csredis.Set("string-key", "hello ");

// 在指定key的value末尾追加字符串
csredis.Append("string-key", "world"); 
// output -> "hello world"

// 获取从指定范围所有字符构成的子串（start:3,end:7）
csredis.GetRange("string-key",3,7)  
// output ->  "lo wo"
    
// 用新字符串从指定位置覆写原value（index:4）
csredis.SetRange("string-key", 4, "aa"); 
// output -> "hellaaword"





//-----------------列表(list)----------------
// 从右端推入元素
csredis.RPush("my-list", "item1", "item2", "item3", "item4"); 
// 从右端弹出元素
csredis.RPop("my-list");
// 从左端推入元素
csredis.LPush("my-list","LeftPushItem");
// 从左端弹出元素
csredis.LPop("my-list");

// 遍历链表元素（start:0,end:-1即可返回所有元素）
foreach (var item in csredis.LRange("my-list", 0, -1))
{
    Console.WriteLine(item);
}
// 按索引值获取元素（当索引值大于链表长度，返回空值，不会报错）
Console.WriteLine($"{csredis.LIndex("my-list", 1)}"); 

// 修剪指定范围内的元素（start:4,end:10）
csredis.LTrim("my-list", 4, 10);

// 将my-list最后一个元素弹出并压入another-list的头部
csredis.RPopLPush("my-list", "another-list");





//------------------集合(set)----------------
// 实际上只插入了两个元素("item1","item2")
csredis.SAdd("my-set", "item1", "item1", "item2"); 

// 集合的遍历
foreach (var member in csredis.SMembers("my-set"))
{
    Console.WriteLine($"集合成员：{member.ToString()}");
}

// 判断元素是否存在
string member = "item1";
Console.WriteLine($"{member}是否存在:{csredis.SIsMember("my-set", member)}"); 
// output -> True

// 移除元素
csredis.SRem("my-set", member);
Console.WriteLine($"{member}是否存在:{csredis.SIsMember("my-set", member)}"); 
// output ->  False

// 随机移除一个元素
csredis.SPop("my-set");

csredis.SAdd("set-a", "item1", "item2", "item3","item4","item5");
csredis.SAdd("set-b", "item2", "item5", "item6", "item7");

// 差集
csredis.SDiff("set-a", "set-b"); 
// output -> "item1", "item3","item4"

// 交集
csredis.SInter("set-a", "set-b"); 
// output -> "item2","item5"

// 并集
csredis.SUnion("set-a", "set-b");
// output -> "item1","item2","item3","item4","item5","item6","item7"







//------------------散列(hashmap)----------------
// 向散列添加元素
csredis.HSet("ArticleID:10001", "Title", "了解简单的Redis数据结构");
csredis.HSet("ArticleID:10001", "Author", "xscape");
csredis.HSet("ArticleID:10001", "PublishTime", "2019-01-01");

// 根据Key获取散列中的元素
csredis.HGet("ArticleID:10001", "Title");

// 获取散列中的所有元素
foreach (var item in csredis.HGetAll("ArticleID:10001"))
{
    Console.WriteLine(item.Value);
}

//HMGet和HMSet是他们的多参数版本，一次可以处理多个键值对
var keys = new string[] { "Title","Author","publishTime"};
csredis.HMGet("ID:10001", keys);

//和处理字符串一样，我们也可以对散列中的值进行自增、自减操作，原理同字符串是一样的
csredis.HSet("ArticleID:10001", "votes", "257");
csredis.HIncrBy("ID:10001", "votes", 40);
// output -> 297




//------------------有序集合----------------
// 向有序集合添加元素
csredis.ZAdd("Quiz", (79, "Math"));
csredis.ZAdd("Quiz", (98, "English"));
csredis.ZAdd("Quiz", (87, "Algorithm"));
csredis.ZAdd("Quiz", (84, "Database"));
csredis.ZAdd("Quiz", (59, "Operation System"));

//返回集合中的元素数量
csredis.ZCard("Quiz");

// 获取集合中指定范围(90~100)的元素集合
csredis.ZRangeByScore("Quiz",90,100);

// 获取集合所有元素并升序排序
csredis.ZRangeWithScores("Quiz", 0, -1);

// 移除集合中的元素
csredis.ZRem("Quiz", "Math");

//Key的过期
redis.Set("MyKey", "hello,world");
Console.WriteLine(redis.Get("MyKey")); 
// output -> "hello,world"

redis.Expire("MyKey", 5); // key在5秒后过期，也可以使用ExpireAt方法让它在指定时间自动过期

Thread.Sleep(6000); // 线程暂停6秒
Console.WriteLine(redis.Get("MyKey"));
// output -> ""

https://www.cnblogs.com/xscape/p/10208638.html

3.高级玩法：发布订阅

//普通订阅
rds.Subscribe(
  ("chan1", msg => Console.WriteLine(msg.Body)),
  ("chan2", msg => Console.WriteLine(msg.Body)));

//模式订阅（通配符）
rds.PSubscribe(new[] { "test*", "*test001", "test*002" }, msg => {
  Console.WriteLine($"PSUB   {msg.MessageId}:{msg.Body}    {msg.Pattern}: chan:{msg.Channel}");
});
//模式订阅已经解决的难题：
//1、分区的节点匹配规则，导致通配符最大可能匹配全部节点，所以全部节点都要订阅
//2、本组 "test*", "*test001", "test*002" 订阅全部节点时，需要解决同一条消息不可执行多次

//发布
rds.Publish("chan1", "123123123");
//无论是分区或普通模式，rds.Publish 都可以正常通信

4.高级玩法：缓存壳

//不加缓存的时候，要从数据库查询
var t1 = Test.Select.WhereId(1).ToOne();

//一般的缓存代码，如不封装还挺繁琐的
var cacheValue = rds.Get("test1");
if (!string.IsNullOrEmpty(cacheValue)) {
    try {
        return JsonConvert.DeserializeObject(cacheValue);
    } catch {
        //出错时删除key
        rds.Remove("test1");
        throw;
    }
}
var t1 = Test.Select.WhereId(1).ToOne();
rds.Set("test1", JsonConvert.SerializeObject(t1), 10); //缓存10秒

//使用缓存壳效果同上，以下示例使用 string 和 hash 缓存数据
var t1 = rds.CacheShell("test1", 10, () => Test.Select.WhereId(1).ToOne());
var t2 = rds.CacheShell("test", "1", 10, () => Test.Select.WhereId(1).ToOne());
var t3 = rds.CacheShell("test", new [] { "1", "2" }, 10, notCacheFields => new [] {
  ("1", Test.Select.WhereId(1).ToOne()),
  ("2", Test.Select.WhereId(2).ToOne())
});

5.高级玩法：管道

使用管道模式，打包多条命令一起执行，从而提高性能。

var ret1 = rds.StartPipe().Set("a", "1").Get("a").EndPipe();
var ret2 = rds.StartPipe(p => p.Set("a", "1").Get("a"));

var ret3 = rds.StartPipe().Get("b").Get("a").Get("a").EndPipe();
//与 rds.MGet("b", "a", "a") 性能相比，经测试差之毫厘

6.高级玩法：多数据库

var connectionString = "127.0.0.1:6379,password=123,poolsize=10,ssl=false,writeBuffer=10240,prefix=key前辍";
var redis = new CSRedisClient[14]; //定义成单例
for (var a = 0; a< redis.Length; a++) redis[a] = new CSRedisClient(connectionString + "; defualtDatabase=" + a);

//访问数据库1的数据
redis[1].Get("test1");

7.性能比拼

Redis的五种数据类型的简单介绍和使用 https://www.cnblogs.com/dddyyy/p/9803828.html

来源： .net 5.0 – 使用CSRedisCore操作redis – gygtech – 博客园

ServiceStack.Redis 是商业版，免费版有限制；

StackExchange.Redis 是免费版，但是内核在 .NETCore 运行有问题经常 Timeout，暂无法解决；

CSRedis于2016年开始支持.NETCore一直迭代至今，实现了低门槛、高性能，和分区高级玩法的.NETCore redis-cli SDK；

在v3.0版本更新中，CSRedis中的所有方法名称进行了调整，使其和redis-cli保持一致，如果你熟悉redis-cli的命令的话，CSRedis可以直接上手，这样学习成本就降低很多。

•  引用包

•  appsettings.json

•  初始化

•  StartUp类配置

127.0.0.1:6379,password=YourPassword,defaultDatabase=0,prefix=hr_

• 简单操作

• 对整数类型进行自增，自减操作

• 在指定 key 的 value 末尾追加字符串

• #HSet、HGet、HDel方法 [只能处理一个键值对]

• #HGetAll、HKeys、HVals

• #HMSet、HMGet [HGet和HSet方法执行一次只能处理一个键值对，而HMGet和HMSet是他们的多参数版本，一次可以处理多个键值对。]

• #对散列中的值进行自增、自减操作

• #对集合中的成员进行操作

• #对两个集合进行交、并、差操作

• 有序集合可以看作是可排序的散列，不过有序集合的val成为score分值，集合内的元素就是基于score进行排序的，score以双精度浮点数的格式存储。

来源： Redis ZADD 命令

Redis ZADD 命令用于将一个或多个 member 元素及其 score 值加入到有序集 key 当中。

如果某个 member 已经是有序集的成员，那么更新这个 member 的 score 值，并通过重新插入这个 member 元素，来保证该 member 在正确的位置上。

如果有序集合 key 不存在，则创建一个空的有序集并执行 ZADD 操作。

当 key 存在但不是有序集类型时，返回一个错误。

score 值可以是整数值或双精度浮点数，score 可为正也可以为负。

对有序集的更多介绍请参见 sorted set 。

注意： 在 Redis 2.4 版本以前， ZADD 每次只能添加一个元素。

语法

redis ZADD 命令基本语法如下：

redis 127.0.0.1:6379> ZADD key [NX|XX] [CH] [INCR] score member [score member …]

添加单个元素

redis> ZADD page_rank 10 google.com
(integer) 1

添加多个元素

redis> ZADD page_rank 9 baidu.com 8 redis.com.cn
(integer) 2

redis> ZRANGE page_rank 0 -1 WITHSCORES
1) "redis.com.cn"
2) "8"
3) "baidu.com"
4) "9"
5) "google.com"
6) "10"

添加已存在元素，且 score 值不变

redis> ZADD page_rank 10 google.com
(integer) 0

redis> ZRANGE page_rank 0 -1 WITHSCORES  # 没有改变
1) "redis.com.cn"
2) "8"
3) "baidu.com"
4) "9"
5) "google.com"
6) "10"

添加已存在元素，但是改变 score 值

redis> ZADD page_rank 6 redis.com.cn
(integer) 0

redis> ZRANGE page_rank 0 -1 WITHSCORES  # redis.com.cn 元素的 score 值被改变
1) "redis.com.cn"
2) "6"
3) "baidu.com"
4) "9"
5) "google.com"
6) "10"

ZADD 参数

ZADD 支持参数，参数位于 key 名字和第一个 score 参数之间:

• XX: 仅更新存在的成员，不添加新成员。
• NX: 不更新存在的成员。只添加新成员。
• LT: 更新新的分值比当前分值小的成员，不存在则新增。
• GT: 更新新的分值比当前分值大的成员，不存在则新增。
• CH: 返回变更成员的数量。变更的成员是指 新增成员 和 score值更新的成员，命令指明的和之前score值相同的成员不计在内。 注意: 在通常情况下，ZADD返回值只计算新添加成员的数量。
• INCR: ZADD 使用该参数与 ZINCRBY 功能一样。一次只能操作一个score-element对。

注意: GT, LT 和 NX 三者互斥不能同时使用。

scores 有效值范围

Redis 有序集合的分数使用双精度64位浮点数表示。在Redis所支持的平台上，称为IEEE 754 floating point number，它能包括的整数范围是-(2^53) 到 +(2^53)。或者说是-9007199254740992 到 9007199254740992。更大的整数在内部用指数形式表示，所以，如果为分数设置一个非常大的整数，你得到的是一个近似的十进制数。

Sorted sets 101

有序集合按照分数以递增的方式进行排序。相同的成员（member）只存在一次，有序集合不允许存在重复的成员。 分数可以通过ZADD命令进行更新或者也可以通过ZINCRBY命令递增来修改之前的值，相应的他们的排序位置也会随着分数变化而改变。

获取一个成员当前的分数可以使用 ZSCORE 命令，也可以用它来验证成员是否存在。

更多关于有序集合的信息请参考 sorted sets.

相同分数的成员

有序集合里面的成员是不能重复的都是唯一的，但是，不同成员间有可能有相同的分数。当多个成员有相同的分数时，他们将是按字典排序（ordered lexicographically）（仍由分数作为第一排序条件，然后，相同分数的成员按照字典序排序）。

字典顺序排序用的是二进制，它比较的是字符串的字节数组。

如果用户将所有元素设置相同分数（例如0），有序集合里面的所有元素将按照字典顺序进行排序，范围查询元素可以使用 ZRANGEBYLEX 命令（注：范围查询分数可以使用 ZRANGEBYSCORE 命令）。

返回值

整数:

• 被成功添加的新成员的数量，不包括那些被更新分数的、已经存在的成员。

如果使用 INCR 选项，则返回 多行字符串:

• 以字符串形式表示的 member 的 score 值(双精度浮点数)。(双精度浮点数) , 执行失败返回 nil (当使用 XX 或 NX 选项)。

历史

• >= 2.4: 支持一次增加或更新多个成员。
• >= 3.0.2: 增加 XX, NX, CH 和 INCR 选项。
• >=6.2: 增加 GT 和 LT 选项。

例子

(integer) 1

(integer) 1

(integer) 2

1) "one"
2) "1"
3) "uno"
4) "1"
5) "two"
6) "2"
7) "three"
8) "3"

来源： MSMQ 的持久化_走错路的程序员的博客-CSDN博客

MSMQ的消息默认是放在内存里面的. 重启服务或者断电的时候消息就没了. 对于重要的消息来讲这样肯定是不行的.

百度了好久也没发现如何持久化. 后来实在不行上Google 才找到对应的答案.
废话少说. 直接上答案.

标准答案就一句话 msg1.Recoverable = true;

System.Messaging.Message msg1 = new System.Messaging.Message(person);
msg1.Recoverable = true;//为true 的时候进行持久化.每个信息的Recoverable 为true 那么持久保存
mq.Send(msg1, transaction);

//下面的是全部的测试代码.. 可以不看..
//主测试方法

public class TestMQ
{

static string path = “.\\private$\\TestMQ”;
// 接收消息的函数
public static Object[] Receive()
{
MessageQueue mq;
if (MessageQueue.Exists(path))
{
mq = new MessageQueue(path);
}
else
{
mq = MessageQueue.Create(path);
}

// 定义消息队列中所有的消息种类（种类之间的顺序可以互换）
Type[] msgTypes = new Type[] { typeof(Person), typeof(Order), typeof(String) };

// 规定消息以XML格式编码
mq.Formatter = new XmlMessageFormatter(msgTypes);

// 定义事务
MessageQueueTransaction transaction = new MessageQueueTransaction();

try
{
// 如果消息队列采用了事务，则开始事务
if (mq.Transactional)
transaction.Begin();

System.Messaging.Message msg1 = mq.Receive(transaction);
Person person = (Person)msg1.Body;

System.Messaging.Message msg2 = mq.Receive(transaction);
Order order = (Order)msg2.Body;

System.Messaging.Message msg3 = mq.Receive(transaction);
string description = (String)msg3.Body;

// 如果消息队列采用了事务，则停止事务
if (mq.Transactional)
transaction.Commit();

return new Object[] { person, order, description };
}
catch (Exception ex)
{
// 如果消息队列采用了事务并且出现了异常，则终止事务
if (mq.Transactional)
transaction.Abort();

return null;
}
}

// 发送消息的函数
public static bool Send(Person person, Order order, string description)
{
MessageQueue mq;
if (MessageQueue.Exists(path))
{
mq = new MessageQueue(path);
}
else
{
mq = MessageQueue.Create(path);
}
mq.DefaultPropertiesToSend.Recoverable = true; //为true 的时候进行持久化. 经测试好像没用.
//mq.DefaultPropertiesToSend.AcknowledgeType = AcknowledgeTypes.NegativeReceive; 消息是否需要反馈.
// 规定消息以XML格式编码
mq.Formatter = new XmlMessageFormatter(new Type[] { typeof(Person), typeof(Order), typeof(String) });

// 定义事务
MessageQueueTransaction transaction = new MessageQueueTransaction();
System.Messaging.Message msg1 = new System.Messaging.Message(person);
System.Messaging.Message msg2 = new System.Messaging.Message(order);
System.Messaging.Message msg3 = new System.Messaging.Message(description);

try
{
// 如果消息队列采用了事务，则开始事务
if (mq.Transactional)
transaction.Begin();

msg1.Recoverable = true;//为true 的时候进行持久化.
mq.Send(msg1, transaction);

msg2.Recoverable = true;//为true 的时候进行持久化.
mq.Send(msg2, transaction);

msg3.Recoverable = false;//为true 的时候进行持久化.
mq.Send(msg3, transaction);

// 如果消息队列采用了事务，则停止事务
if (mq.Transactional) {
transaction.Commit();
}
return true;
}
catch (Exception ex)
{
// 如果消息队列采用了事务并且出现了异常，则终止事务
if (mq.Transactional)
transaction.Abort();

return false;
}
finally
{

mq.Close();
msg1 = null;
msg2 = null;
msg3 = null;
}
}

}

下面是两个model类

// 购物清单
public class Order
{
public Order()
{
Price = 0.0;
Number = 0;
Time = DateTime.Now;
}

public Order(Double price, UInt64 number, DateTime time)
{
Price = price;
Number = number;
Time = time;
}

// 物品单价
Double Price;

// 物品数量
UInt64 Number;

// 下单时间
DateTime Time;
}

// 人员类
public class Person
{
public Person()
{
m_Name = “”;
m_Age = 0;
}

public Person(string name, UInt16 age)
{
m_Name = name;
m_Age = age;
}

// 姓名
public string Name
{
get { return m_Name; }
set { m_Name = value; }
}

// 年龄
public UInt16 Age
{
get { return m_Age; }
set { m_Age = value; }
}

private string m_Name;
private UInt16 m_Age;
}

//发送按钮
private void 发送按钮_Click(object sender, EventArgs e)
{

for (int i = 0; i < 100000; i++)
{
Person person = new Person(“Jackie”, 30);
Order order = new Order(110.0, 10, DateTime.Now);
string description = “This is a new order.”;
TestMQ.Send(person, order, description);
System.GC.Collect();
}

}

private void 接收按钮_Click(object sender, EventArgs e)
{
// 接收消息。返回值为 null 时表示接收失败
Object[] message = TestMQ.Receive();

if (message != null)
{
// 当消息接收成功时，将消息依次取出并存入下面的对象中
Person person = (Person)message[0];
Order prder = (Order)message[1];
string description = (string)message[2];
if (description == “This is a new order.”)
{
MessageBox.Show(“消息接收成功 !”);
}
else
{
MessageBox.Show(“消息接收失败 !”);
}

}
else
{
MessageBox.Show(“消息接收失败 !”);
}
}
————————————————
版权声明：本文为CSDN博主「走错路的程序员」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/phker/article/details/71124256