排序 #

排序#

初级排序算法#

选择排序#

命题A。对于长度为N 的数组，选择排序需要大约 N^2/2 次比较和N 次交换。

代码

public class Selection
{
	public static void sort(Comparable[] a)
	{
		// 将a[]按升序排列
		int N = a.length;
		// 数组长度
		for (int i = 0; i < N; i++)
		{
			// 将a[i]和a[i+1..N]中最小的元素交换
			int min = i;
			// 最小元素的索引
			for (int j = i+1; j < N; j++)
			if (less(a[j], a[min])) min = j;
			exch(a, i, min);
		}
	}
	// less()、exch()、isSorted()和main()方法见“排序算法类模板”
}

特点
- 运行时间与输入无关，即输入数据的初始状态（比如是否已排序好等等）不影响排序时间
- 数据移动是最少的（只使用了N次交换，交换次数和数组的大小是线性关系

插入排序#

命题B。对于随机排列的长度为N 且主键不重复的数组，平均情况下插入排序需要～ N^{2/4 次比较以及～ N}2/4 次交换。最坏情况下需要～ N^{2/2 次比较和～ N}2/2 次交换，最好情况下需要N-1次比较和0 次交换。

算法红皮书 2.1.1

2022年4月13日 22:43 周三

排序#

排序就是将一组对象按照某种逻辑顺序重新排序的过程

对排序算法的分析有助于理解本书中比较算法性能的方法
类似技术能解决其他类型问题
排序算法常常是我们解决其他问题的第一步

初级排序算法#

熟悉术语及技巧
某些情况下初级算法更有效
有助于改进复杂算法的效率

游戏规则#

主要关注重新排序数组元素的算法，每个元素都会有一个主键
排序后索引较大的主键大于索引较小的主键
一般情况下排序算法通过两个方法操作数据，less()进行比较，exch()进行交换

排序算法类的模板

public class Example
{
	public static void sort(Comparable[] a)
	{
		/* 请见算法2.1、算法2.2、算法2.3、算法2.4、算法2.5或算法2.7*/
	}
	private static Boolean less(Comparable v, Comparable w)
	{
		return v.compareTo(w) < 0;
	}
	private static void exch(Comparable[] a, int i, int j)
	{
		Comparable t = a[i];
		a[i] = a[j];
		a[j] = t;
	}
	private static void show(Comparable[] a)
	{
		// 在单行中打印数组
		for (int i = 0; i < a.length; i++)
		StdOut.print(a[i] + " ");
		StdOut.println();
	}
	public static Boolean isSorted(Comparable[] a)
	{
		// 测试数组元素是否有序
		for (int i = 1; i < a.length; i++)
		if (less(a[i], a[i-1])) return false;
		return true;
	}
	public static void main(String[]
	args)
	{
		// 从标准输入读取字符串，将它们排序并输出
		String[] a = In.readStrings();
		sort(a);
		assert isSorted(a);
		show(a);
	}
}

使用

05高级功能

2022年4月9日 09:20 周六

学习来源 https://www.bilibili.com/video/BV1L4411y7mn（添加小部分笔记）感谢作者！

消息存储#

流程#

ly-20241212142201154

存储介质#

关系型数据库DB#

适合数据量不够大，比如ActiveMQ可选用JDBC方式作为消息持久化

文件系统#

关系型数据库最终也是要存到文件系统中的，不如直接存到文件系统，绕过关系型数据库
常见的RocketMQ/RabbitMQ/Kafka都是采用消息刷盘到计算机的文件系统来做持久化(同步刷盘/异步刷盘)

消息发送#

顺序写：600MB/s，随机写：100KB/s
- 系统运行一段时间后，我们对文件的增删改会导致磁盘上数据无法连续，非常的分散。
- 顺序读也只是逻辑上的顺序，也就是按照当前文件的相对偏移量顺序读取，并非磁盘上连续空间读取
- 对于磁盘的读写分为两种模式，顺序IO和随机IO。随机IO存在一个寻址的过程，所以效率比较低。而顺序IO，相当于有一个物理索引，在读取的时候不需要寻找地址，效率很高。
- 来源： https://www.cnblogs.com/liuche/p/15455808.html
数据网络传输
零拷贝技术MappedByteBuffer，省去了用户态，由内核态直接拷贝到网络驱动内核。
RocketMQ默认设置单个CommitLog日志数据文件为1G

消息存储#

三个概念：commitLog、ConsumerQueue、index

CommitLog#

默认大小1G
存储消息的元数据，包括了Topic、QueueId、Message
还存储了ConsumerQueue相关信息，所以ConsumerQueue丢了也没事

ConsumerQueue#

存储了消息在CommitLog的索引（几百K，Linux会事先加载到内存中）
包括最小/最大偏移量、已经消费的偏移量
一个Topic多个队列，每个队列对应一个ConsumerQueue

Index#

也是索引文件，为消息查询服务，通过key或时间区间查询消息

总结#

ly-20241212142201312

刷盘机制#

ly-20241212142201340

同步刷盘
异步刷盘

高可用性机制#

消费高可用及发送高可用#

ly-20241212142201372

消息主从复制#

ly-20241212142201400

负载均衡#

ly-20241212142201467

消息重试#

下面都是针对消费失败的重试

顺序消息#

RocketMQ会自动不断重试，且为了保证顺序性，会导致消息消费被阻塞。使用时要及时监控并处理消费失败现象

无序消息（普通、定时、延时、事务）#

通过设置返回状态达到消息重试的结果
重试只对集群消费方式生效，广播方式不提供重试特性
重试次数如果16次后还是消费失败，会进入死信队列，不再被消费

配置是否重试#

重试#

ly-20241212142201525

算法红皮书 1.5.1-1.5.3

2022年4月9日 08:01 周六

案例研究：union-find 算法#

设计和分析算法的基本方法
- 优秀的算法能解决实际问题
- 高效的算法也可以很简单
- 理解某个实现的性能特点是一项有趣的挑战
- 在解决同一个问题的多种算法间选择，科学方法是一种重要工具
- 迭代式改进能让算法效率越来越高

动态连通性#

从输入中读取整数对p q，如果已知的所有整数对都不能说明p,q相连，就打印出pq
网络：整个程序能够判定是否需要在pq之间架设一条新的连接才能进行通信
变量名等价性（即指向同一个对象的多个引用）
数学集合：在处理一个整数对pq时，我们是在判断它们是否属于相同的集合
本节中，将对象称为触点，整数对称为连接，等价类称为连通分量或是简称分量
连通性问题只要求我们的程序能够判别给定的整数对pq是否相连，并没有要求给两者之间的通路上的所有连接
union-find算法的API
数据结构和算法的设计影响到算法的效率

实现#

public class UF
{
	private int[]	id;
	/* 分量id（以触点作为索引） */
	private int	count;
	/* 分量数量 */
	public UF( int N )
		{
		/* 初始化分量id数组 */
		count	= N;
		id	= new int[N];
		for ( int i = 0; i < N; i++ )
					id[i] = i;
	}
	public int count()
		{
		return(count);
	}
	public Boolean connected( int p, int q )
		{
		return(find( p ) == find( q ) );
	}
	public int find( int p )
		public void union( int p, int q )
	/* 请见1.5.2.1节用例（quick-find）、1.5.2.3节用例（quick-union）和算法1.5（加权quick-union） */
	public static void main( String[] args )
		{
		/* 解决由StdIn得到的动态连通性问题 */
		int	N	= StdIn.readint();
		/* 读取触点数量 */
		UF	uf	= new UF( N );
		/* 初始化N个分量 */
		while ( !StdIn.isEmpty() )
				{
			int	p	= StdIn.readint();
			int	q	= StdIn.readint();
			/* 读取整数对 */
			if ( uf.connected( p, q ) )
							continue;
			/* 如果已经连通则忽略 */
			uf.union( p, q );
			/* 归并分量 */
			StdOut.println( p + " " + q );
			/* 打印连接 */
		}
		StdOut.println( uf.count() + "components" );
	}
}

union-find的成本模型：union-find API的各种算法，统计的是数组的访问次数，不论读写

04案例

2022年4月8日 11:00 周五

学习来源 https://www.bilibili.com/video/BV1L4411y7mn（添加小部分笔记）感谢作者!基本架构

架构#

ly-20241212142200977

流程图#

下单流程#

ly-20241212142201029

支付流程#

ly-20241212142201068

SpringBoot整合RocketMQ#

依赖包#

	   <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
        </dependency>

生产者#

yaml#

rocketmq:
  name-server: 192.168.1.135:9876;192.168.1.138:9876
  producer:
    group: my-group

使用#

    @Autowired
    private RocketMQTemplate template;

    @RequestMapping("rocketmq")
    public String rocketmq(){
        log.info("我被调用了-rocketmq");
        //主题+内容
        template.convertAndSend("mytopic-ly","hello1231");
        return "hello world"+serverPort;
    }

消费者#

yaml#

rocketmq:
  name-server: 192.168.1.135:9876;192.168.1.138:9876
  consumer:
    group: my-group2

使用#

创建监听器

@RocketMQMessageListener(topic = "mytopic-ly",
        consumeMode = ConsumeMode.CONCURRENTLY,consumerGroup = "${rocketmq.producer.group}")
@Slf4j
@Component
public class Consumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String s) {
        log.info("消费了"+s);
    }
}

下单流程利用MQ进行回退处理，保证数据一致性#

ly-20241212142201106

库存回退的消费者，代码如下：

@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup =
 "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{


    @Value("${mq.order.consumer.group.name}")
    private String groupName;

    @Autowired
    private TradeGoodsMapper goodsMapper;

    @Autowired
    private TradeMqConsumerLogMapper mqConsumerLogMapper;

    @Autowired
    private TradeGoodsNumberLogMapper goodsNumberLogMapper;

    @Override
    public void onMessage(MessageExt messageExt) {
        String msgId=null;
        String tags=null;
        String keys=null;
        String body=null;
        try {
            //1. 解析消息内容
            msgId = messageExt.getMsgId();
            tags= messageExt.getTags();
            keys= messageExt.getKeys();
            body= new String(messageExt.getBody(),"UTF-8");

            log.info("接受消息成功");

            //2. 查询消息消费记录
            TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
            primaryKey.setMsgTag(tags);
            primaryKey.setMsgKey(keys);
            primaryKey.setGroupName(groupName);
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);

            if(mqConsumerLog!=null){
                //3. 判断如果消费过...
                //3.1 获得消息处理状态
                Integer status = mqConsumerLog.getConsumerStatus();
                //处理过...返回
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().intValue()==status.intValue()){
                    log.info("消息:"+msgId+",已经处理过");
                    return;
                }

                //正在处理...返回
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().intValue()==status.intValue()){
                    log.info("消息:"+msgId+",正在处理");
                    return;
                }

                //处理失败
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue()==status.intValue()){
                    //获得消息处理次数
                    Integer times = mqConsumerLog.getConsumerTimes();
                    if(times>3){
                        log.info("消息:"+msgId+",消息处理超过3次,不能再进行处理了");
                        return;
                    }
                    mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());

                    //使用数据库乐观锁更新
                    TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
                    TradeMqConsumerLogExample.Criteria criteria = example.createCriteria();
                    criteria.andMsgTagEqualTo(mqConsumerLog.getMsgTag());
                    criteria.andMsgKeyEqualTo(mqConsumerLog.getMsgKey());
                    criteria.andGroupNameEqualTo(groupName);
                    criteria.andConsumerTimesEqualTo(mqConsumerLog.getConsumerTimes());
                    int r = mqConsumerLogMapper.updateByExampleSelective(mqConsumerLog, example);
                    if(r<=0){
                        //未修改成功,其他线程并发修改
                        log.info("并发修改,稍后处理");
                    }
                }

            }else{
                //4. 判断如果没有消费过...
                mqConsumerLog = new TradeMqConsumerLog();
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                mqConsumerLog.setGroupName(groupName);
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                mqConsumerLog.setConsumerTimes(0);

                //将消息处理信息添加到数据库
                mqConsumerLogMapper.insert(mqConsumerLog);
            }
            //5. 回退库存
            MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
            Long goodsId = mqEntity.getGoodsId();
            TradeGoods goods = goodsMapper.selectByPrimaryKey(goodsId);
            goods.setGoodsNumber(goods.getGoodsNumber()+mqEntity.getGoodsNum());
            goodsMapper.updateByPrimaryKey(goods);


            //6. 将消息的处理状态改为成功
            mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode());
            mqConsumerLog.setConsumerTimestamp(new Date());
            mqConsumerLogMapper.updateByPrimaryKey(mqConsumerLog);
            log.info("回退库存成功");
        } catch (Exception e) {
            e.printStackTrace();
            TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
            primaryKey.setMsgTag(tags);
            primaryKey.setMsgKey(keys);
            primaryKey.setGroupName(groupName);
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);
            if(mqConsumerLog==null){
                //数据库未有记录
                mqConsumerLog = new TradeMqConsumerLog();
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                mqConsumerLog.setGroupName(groupName);
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                mqConsumerLog.setConsumerTimes(1);
                mqConsumerLogMapper.insert(mqConsumerLog);
            }else{
                mqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes()+1);
                mqConsumerLogMapper.updateByPrimaryKeySelective(mqConsumerLog);
            }
        }

    }
}

03收发消息

2022年4月7日 14:31 周四

学习来源 https://www.bilibili.com/video/BV1L4411y7mn（添加小部分笔记）感谢作者!前提

依赖包#

		<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.4.0</version>
        </dependency>

消息生产者步骤#

创建生产者，生产者组名–>指定nameserver地址–>启动producer–>

创建消息对象(Topic、Tag、消息体)

发送消息、关闭生产者producer

消息消费者步骤#

创建消费者，制定消费者组名–>指定nameserver地址

订阅Topic和Tag，设置回调函数处理消息

启动消费者consumer

消息发送#

同步消息#

发送消息后客户端会进行阻塞，直到得到结果后，客户端才会继续执行

    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        //创建Producer，并指定生产者组
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.1.135:9876;192.168.1.138:9876");
        producer.start();
        for (int i = 0; i < 10; i++) {
            Message msg = new Message();
            msg.setTopic("base");
            msg.setTags("Tag1");
            msg.setBody(("hello world" + i).getBytes());
            //发送消息
            SendResult result = producer.send(msg);
            //发送状态
            SendStatus sendStatus = result.getSendStatus();
            //消息id
            String msgId = result.getMsgId();
            //消息接收队列id
            MessageQueue messageQueue = result.getMessageQueue();
            int queueId = messageQueue.getQueueId();
            log.info(result.toString());
            log.info(messageQueue.toString());
            log.info("status:" + sendStatus +
                    "msgId:" + msgId + "queueId" + queueId);
            TimeUnit.SECONDS.sleep(1);
        }
        log.info("发送结束===================");
        producer.shutdown();
    }

异步消息#

发送消息后不会导致阻塞，当broker返回结果时，会调用回调函数进行处理

算法红皮书 1.4.1-1.4.10

2022年4月7日 11:00 周四

算法分析#

使用数学分析为算法成本建立简洁的模型，并使用实验数据验证这些模型

科学方法#

观察、假设、预测、观察并核实预测、反复确认预测和观察
原则：实验可重现

观察#

计算性任务的困难程度可以用问题的规模来衡量
问题规模可以是输入的大小或某个命令行参数的值
研究问题规模和运行时间的关系

使用计时器得到大概的运行时间 ly-20241212142054451

典型用例

public static void main(String[] args) {
        int N = Integer.parseInt(args[0]);
        int[] a = new int[N];
        for (int i = 0; i < N; i++)
            a[i] = StdRandom.uniform(-1000000, 1000000);
        Stopwatch timer = new Stopwatch();
        int cnt = ThreeSum.count(a);
        double time = timer.elapsedTime();
        StdOut.println(cnt + " triples " + time + " seconds");
    }

使用方法

数据类型的实现

public class Stopwatch {
    private final long start;

    public Stopwatch() {
        start = System.currentTimeMillis();
    }

    public double elapsedTime() {
        long now = System.currentTimeMillis();

        return (now - start) / 1000.0;
    }
}

数学模型#

程序运行的总时间主要和两点有关：执行每条语句的耗时；执行每条语句的频率

算法红皮书1.3.3.1-1.3.4

2022年4月6日 23:00 周三

背包、队列和栈#

链表#

链表是一种递归的数据结构，它或者为空(null)，或者是一个指向一个结点（node）的引用，该节点含有一个泛型的元素和一个指向另一条链表的引用。

结点记录#

使用嵌套类定义结点的抽象数据类型
```
private class Node
{
	Item item;
	Node next;
}
```
- 该类没有其它任何方法，且会在代码中直接引用实例变量，这种类型的变量称为记录

构造链表#

需要一个Node类型的变量，保证它的值是null或者指向另一个Node对象的next域指向了另一个链表
如下图
链表表示的是一列元素
链式结构在本书中的可视化表示长方形表示对象；实例变量的值写在长方形中；用指向被引用对象的箭头表示引用关系
术语链接表示对结点的引用

在表头插入结点#

在首结点为first 的给定链表开头插入字符串not，我们先将first 保存在oldfirst 中，然后将一个新结点赋予first，并将它的item 域设为not，next 域设为oldfirst
时间复杂度为O(1)
如图

从表头删除结点#

将first指向first.next
原先的结点称为孤儿，Java的内存管理系统最终将回收它所占用的内存
如图

在表尾插入结点#

每个修改链表的操作都需要增加检查是否要修改该变量（以及做出相应修改）的代码
例如，当删除链表首结点时可能改变指向链表的尾结点的引用，因为链表中只有一个结点时它既是首结点又是尾结点
如图

其他位置的插入和删除操作#

删除指定结点；在指定节点插入新结点

需要将链表尾结点的前一个节点中的链接（它指向的是last）值改为null
为了找到指向last的结点，需要遍历链表，时间复杂度为O(n)
实现任意插入和删除操作的标准解决方案是双向链表

遍历#

将x初始化为链表首结点，然后通过x.item访问和x相关联的元素，并将x设为x.next来访问链表中的下一个结点，知道x=null(没有下一个结点了，到达链表结尾)
```
for (Node x = first; x != null; x = x.next)
{
// 处理x.item
}
```

栈的实现#

使用链表实现栈
将栈保存为一条链表，栈的顶部即为表头，实例变量first 指向栈顶。这样，当使用push() 压入一个元素时，我们会按照1.3.3.3 节所讨论的代码将该元素添加在表头；当使用pop() 删除一个元素时，我们会按照1.3.3.4 节讨论的代码将该元素从表头删除。要实现size() 方法，我们用实例变量N 保存元素的个数，在压入元素时将N 加1，在弹出元素时将N 减1。要实现isEmpty() 方法，只需检查first 是否为null（或者可以检查N 是否为0）

02双主双从集群搭建

2022年4月6日 10:55 周三